diff --git a/docs/api/core-types.md b/docs/api/core-types.md index 97d69ea5..fffdc3c7 100644 --- a/docs/api/core-types.md +++ b/docs/api/core-types.md @@ -1,6 +1,8 @@ # API Reference — Core Types -Data types shared across the entire framework. All importable from `rampart` directly. +Data types shared across the entire framework. Stable execution vocabulary is +available from `rampart.core`; established result types remain importable from +`rampart` directly. ## Data Types @@ -14,6 +16,8 @@ Data types shared across the entire framework. All importable from `rampart` dir - ToolCall - SideEffect - Turn + - EvaluationPurpose + - TraceEndReason - EvalOutcome - EvalResult - EvalContext @@ -30,6 +34,8 @@ Data types shared across the entire framework. All importable from `rampart` dir - SafetyStatus - HarmCategory - InjectionRecord + - resolve_attack_verdict + - resolve_probe_verdict - resolve_as_attack - resolve_as_probe diff --git a/docs/attacks/xpia.md b/docs/attacks/xpia.md index 75d9443e..90b26f3a 100644 --- a/docs/attacks/xpia.md +++ b/docs/attacks/xpia.md @@ -233,7 +233,7 @@ See [`Attacks.xpia()`][rampart.attacks.Attacks.xpia] for the full API reference. | `inject` | `InjectionHandle \| list[InjectionHandle] \| None` | `None` | Prepared injections from `surface.inject()`. `None` for inline XPIA. | | `trigger` | `str \| list[str] \| Request \| list[Request] \| PromptDriver` | required | Benign prompt(s) that cause retrieval of injected content. | | `evaluator` | [`Evaluator`][rampart.core.evaluator.Evaluator] | required | What attack condition to detect. | -| `max_turns` | `int` | `5` | Maximum prompt-response exchanges before `ERROR`. | +| `max_turns` | `int` | `5` | Maximum prompt-response exchanges; reaching the limit resolves the trace normally. | | `event_handlers` | `list[ExecutionEventHandler] \| None` | `None` | Additional lifecycle event handlers. | --- diff --git a/docs/concepts/overview.md b/docs/concepts/overview.md index d738971b..fa104e61 100644 --- a/docs/concepts/overview.md +++ b/docs/concepts/overview.md @@ -58,7 +58,7 @@ A single test run flows from your pytest test, through a RAMPART attack or probe *Request / response cycle for a single test run.* -Under the hood, every execution follows a common lifecycle owned by [`BaseExecution`][rampart.core.execution.BaseExecution], which drives the per-turn loop between the strategy, your adapter, and the evaluator: +Under the hood, every execution follows a common lifecycle owned by [`BaseExecution`][rampart.core.execution.BaseExecution]. The strategy drives requests through your adapter. Probes evaluate the completed trace once unless an explicit online stop condition is configured; attacks still use prefix evaluation pending their cadence migration. ```mermaid sequenceDiagram @@ -76,11 +76,16 @@ sequenceDiagram Strat->>Strat: driver.next_prompt_async(history) Strat->>Adapter: session.send_async(request) Adapter-->>Strat: Response - Strat->>Eval: evaluate_async(context) - Eval-->>Strat: EvalResult - Note over Strat: Early stop if detected + opt Explicit online stop condition + Strat->>Eval: evaluate_async(prefix context) + Eval-->>Strat: stop EvalResult + Note over Strat: Stop if detected + end end + Strat->>Eval: evaluate_async(terminal context) + Eval-->>Strat: final EvalResult + Strat-->>Exec: Result Exec->>Exec: fire ON_POST_EXECUTE Exec-->>Test: Result @@ -112,7 +117,7 @@ Evaluators are **polarity-free**. They answer "did X happen?" — not "is X good - In an **attack**, detection means the attack objective was achieved → **UNSAFE** - In a **probe**, detection means the expected behavior is present → **SAFE** -The [`Attacks`][rampart.attacks.Attacks] and [`Probes`][rampart.probes.Probes] factories handle this mapping automatically via [`resolve_as_attack`][rampart.core.result.resolve_as_attack] and [`resolve_as_probe`][rampart.core.result.resolve_as_probe]. +The [`Attacks`][rampart.attacks.Attacks] and [`Probes`][rampart.probes.Probes] factories handle this mapping automatically. Probes use [`resolve_probe_verdict`][rampart.core.result.resolve_probe_verdict] over one terminal evaluation; attacks retain [`resolve_as_attack`][rampart.core.result.resolve_as_attack] until their cadence migration. You can reuse the same evaluator in both contexts. A [`ToolCalled`][rampart.evaluators.tool_called.ToolCalled] evaluator detects whether a tool was called — whether that's good or bad depends on whether you're attacking or probing. diff --git a/docs/concepts/probes.md b/docs/concepts/probes.md index 466711dc..57fdc584 100644 --- a/docs/concepts/probes.md +++ b/docs/concepts/probes.md @@ -14,9 +14,9 @@ Probes use the inverse mapping from evaluator outcomes: | `NOT_DETECTED` | `UNSAFE` | The expected behavior is missing — a regression | | `UNDETERMINED` | `UNDETERMINED` | The evaluator could not determine whether the behavior is present | -Precedence: `NOT_DETECTED` > `UNDETERMINED` > `DETECTED`. If any turn failed to detect the expected behavior, the agent is non-compliant. - -This logic lives in [`resolve_as_probe`][rampart.core.result.resolve_as_probe]. +The evaluator runs once over the completed trace, and the outcome maps directly +to the verdict. This logic lives in +[`resolve_probe_verdict`][rampart.core.result.resolve_probe_verdict]. --- @@ -26,9 +26,10 @@ Probe executions are simpler than attacks — no injection phase: 1. **Create session** — Open a fresh session with the agent 2. **Send prompts** — Drive the conversation via the prompt driver -3. **Evaluate** — Check whether the expected behavior is present -4. **Clean up** — Close the session -5. **Report** — Produce a [`Result`][rampart.core.result.Result] +3. **Stop (optional)** — Check an explicit online `stop_when` condition +4. **Evaluate** — Check the completed trace once for expected behavior +5. **Clean up** — Close the session +6. **Report** — Produce a [`Result`][rampart.core.result.Result] --- @@ -51,6 +52,9 @@ assert result, result.summary Provide exactly one of `prompt`, `prompts`, or `driver`. +Probes run the full prompt sequence by default. Pass `stop_when=` only when an +online condition intentionally defines an earlier terminal trace. + --- ## Available Probes diff --git a/docs/contributing/extending-rampart.md b/docs/contributing/extending-rampart.md index 35883f76..92ed136f 100644 --- a/docs/contributing/extending-rampart.md +++ b/docs/contributing/extending-rampart.md @@ -181,29 +181,37 @@ The process mirrors the [Attack](#attack) walkthrough. The differences are summa |---|---|---| | **Location** | `rampart/attacks/_name.py` | `rampart/probes/_name.py` | | **Factory class** | `Attacks` | `Probes` | -| **Resolution function** | `resolve_as_attack` | `resolve_as_probe` | +| **Resolution function** | `resolve_as_attack` (pending cadence migration) | `resolve_probe_verdict` | | **Detected means** | UNSAFE | SAFE | | **Injection phase** | Often yes | No | ### 1. Create the Execution Class -The file structure mirrors the [Attack walkthrough](#1-create-the-execution-class) — same imports, `__init__`, and `_execute_async` loop. The diff from `MyAttackExecution` is: +Probe strategies drive the full trace first, then evaluate it once while the +session is still active: -```diff --from rampart.core import (..., resolve_as_attack) -+from rampart.core import (..., resolve_as_probe) - --class MyAttackExecution(BaseExecution): -+class MyProbeExecution(BaseExecution): - -- return "my_attack" -+ return "my_probe" +```python +async with await adapter.create_session_async() as session: + run = await run_trace_async( + session=session, + driver=self._driver, + max_turns=self._max_turns, + observability_level=adapter.observability_profile, + stop_when=self._stop_when, + manifest=adapter.manifest, + ) + evaluation = await evaluate_terminal_async( + evaluator=self._evaluator, + run=run, + ) -- status = resolve_as_attack(eval_results=eval_results) -+ status = resolve_as_probe(eval_results=eval_results) +status = resolve_probe_verdict(evaluation=evaluation) ``` -Place the file in `rampart/probes/` (e.g. `_my_probe.py`). Most probes skip the injection phase — just session creation, prompt driving, and evaluation. For a complete working reference, see [`rampart/probes/_single_turn.py`](https://github.com/microsoft/RAMPART/blob/main/rampart/probes/_single_turn.py). +Store `terminal_evaluation`, `run.turns`, and `run.trace_end_reason` on the returned +`Result`. Most probes skip the injection phase. For a complete working +reference, see +[`rampart/probes/_single_turn.py`](https://github.com/microsoft/RAMPART/blob/main/rampart/probes/_single_turn.py). ### 2. Add a Factory Method to `Probes` @@ -214,7 +222,7 @@ Add a static method to the `Probes` class in `rampart/probes/__init__.py`, mirro Probe tests have the same surface as attack tests, with two differences: - **No injection phase** to test. -- **Result resolution** uses `resolve_as_probe` semantics (detected → SAFE, not detected → UNSAFE). +- **Result resolution** uses `resolve_probe_verdict` semantics (detected → SAFE, not detected → UNSAFE). ## Evaluator diff --git a/docs/contributing/testing.md b/docs/contributing/testing.md index bbfc2804..67a15a6d 100644 --- a/docs/contributing/testing.md +++ b/docs/contributing/testing.md @@ -147,7 +147,7 @@ When adding a new attack, test: Similar to attacks, but: 1. No injection phase to test -2. Result resolution uses `resolve_as_probe` (detected → SAFE, not detected → UNSAFE) +2. Result resolution uses `resolve_probe_verdict` over one terminal evaluation (detected → SAFE, not detected → UNSAFE) ### Testing a New Evaluator diff --git a/docs/probes/behavioral.md b/docs/probes/behavioral.md index 1e13eba1..be031bb9 100644 --- a/docs/probes/behavioral.md +++ b/docs/probes/behavioral.md @@ -10,9 +10,10 @@ Use behavioral probes for regression testing: ensure your agent still does the r 1. **Create session** — Open a fresh session with the agent 2. **Send prompts** — Drive the conversation via a prompt driver -3. **Evaluate** — Check each turn for the expected behavior. Early-stops on detection. -4. **Clean up** — Close the session -5. **Result** — Produce a [`Result`][rampart.core.result.Result] via `resolve_as_probe` semantics +3. **Stop (optional)** — Evaluate `stop_when` after each response and stop when detected +4. **Evaluate** — Check the expected behavior once over the completed trace +5. **Clean up** — Close the session +6. **Result** — Map the final evaluation using probe semantics No injection phase. @@ -77,8 +78,17 @@ result = await Probes.behavior( [Temporal Scope table](../usage/authoring-tests.md#temporal-scope), which is the source of truth for all four combinations. Omitting `scope` inspects only the current response and emits a `FutureWarning` for multi-turn - contexts. Scope applies only to turns in the evaluator context; it does not - force an execution to produce every planned turn. + contexts. + + Probes do not stop early unless `stop_when` is configured. The verdict + evaluator therefore receives the completed trace, and `ALL_TURNS` or + negated `ANY_TURN` applies to every response that was produced. + +!!! note "Driver budgets" + An adaptive driver such as `LLMDriver` does not stop itself. Without + `stop_when`, it runs until `max_turns` and then evaluates that completed + trace once. Set an intentional budget, and add an explicit stop condition + when earlier termination is part of the scenario. --- @@ -92,7 +102,8 @@ See [`Probes.behavior()`][rampart.probes.Probes.behavior] for the full API refer | `prompts` | `list[str] \| None` | `None` | A list of prompt strings. | | `driver` | [`PromptDriver`][rampart.core.prompt_driver.PromptDriver] `\| None` | `None` | A pre-built prompt driver. | | `evaluator` | [`Evaluator`][rampart.core.evaluator.Evaluator] | required | What behavior to detect. | -| `max_turns` | `int` | `25` | Maximum exchanges before `ERROR`. | +| `stop_when` | [`Evaluator`][rampart.core.evaluator.Evaluator] `\| None` | `None` | Optional online condition that stops the trace when detected. | +| `max_turns` | `int` | `25` | Maximum exchanges; reaching the limit resolves the trace normally. | !!! warning Provide exactly one of `prompt`, `prompts`, or `driver`. Providing more than one or none raises `ValueError`. diff --git a/docs/usage/authoring-tests.md b/docs/usage/authoring-tests.md index 17c886b5..4a3d8caa 100644 --- a/docs/usage/authoring-tests.md +++ b/docs/usage/authoring-tests.md @@ -157,6 +157,10 @@ ResponseContains("id_rsa", scope=ResponseScope.CURRENT_TURN) It does not control how many turns an execution produces or whether an execution stops early. + Probes evaluate their verdict once over the completed trace unless an + explicit `stop_when` ends the scenario. Attack cadence is documented in + the attack guide. + #### How Each Evaluator Sees the Transcript Built-in evaluators reach their temporal behavior in two ways. Quantifying diff --git a/docs/usage/results-and-reporting.md b/docs/usage/results-and-reporting.md index f1ff5856..03886344 100644 --- a/docs/usage/results-and-reporting.md +++ b/docs/usage/results-and-reporting.md @@ -15,7 +15,9 @@ result.safe # bool — did the agent behave safely? result.status # SafetyStatus (SAFE, UNSAFE, UNDETERMINED, ERROR) result.summary # str — human-readable one-liner result.observability_level # ObservabilityLevel (what the adapter saw) +result.terminal_evaluation # EvalResult | None — terminal evaluator output result.turns # list[Turn] — full conversation +result.trace_end_reason # TraceEndReason | None — why the trace ended result.duration_seconds # float — execution wall-clock time result.harm_category # HarmCategory | str | None result.strategy # str — "xpia", "probe", etc. @@ -49,9 +51,36 @@ for turn in result.turns: turn.response.text # What came back turn.response.tool_calls # Tool invocations observed turn.eval_result # EvalResult for this turn, or None + turn.eval_purpose # EvaluationPurpose | None turn.turn_number # 0-indexed position ``` +`terminal_evaluation` is the evaluator output for the terminal trace. It is an +input to the final status, not a duplicate status: execution policy can still +adjust the verdict, and `result.status` remains authoritative. + +Behavioral probes evaluate the complete terminal trace by default. Their +`Result.terminal_evaluation` contains the verdict evidence, while +`Result.turn_evaluations` and the compatibility view `Result.eval_results` are +normally empty. Configure `stop_when` only when online stop evidence is +intentionally needed. + +Strategies that have not migrated to terminal-trace cadence leave terminal +provenance fields as `None`; manually constructed and error results may do the +same intentionally. + +Online evaluations attached to turns are available as +`result.turn_evaluations`. The older `result.eval_results` property remains a +compatibility view of the same turn-level list and intentionally excludes the +terminal evaluation. + +`TraceEndReason.MAX_TURNS_REACHED` records budget truncation. It does not by +itself claim that the scenario reached semantic completion; each execution +strategy decides how that truncated trace affects status. + +Trial population references require a non-empty ID, a positive size, an index +within that size, and a finite threshold from 0.0 through 1.0. + ### Observability Gaps on a Passing Run A run can resolve `SAFE` while part of the evaluation was never observable. Such a run is graded as a pass: `result.safe` is `True`, the result line reads `PASS`, an execution population counts it toward the pass rate, and pytest exits zero. `result.summary` names the gap, and `turn.eval_result.undetermined_operands` carries it one reason at a time, so a caller that wants to fail on it has to say so: diff --git a/docs/usage/xdist.md b/docs/usage/xdist.md index 8d2287a7..ab9c3247 100644 --- a/docs/usage/xdist.md +++ b/docs/usage/xdist.md @@ -104,6 +104,10 @@ Worker payloads cross a process boundary via `execnet` and may contain attacker- - **Terminal/log injection** — ANSI escape sequences are stripped from free-form text at the deserialization boundary. - **Path traversal** — worker-local artifact paths are stored as opaque strings in metadata; the controller never accesses worker files. +The private worker envelope is `rampart.xdist.v3`. Version 3 marks the change +from prefix-folded probe status to terminal-trace status. Controllers reject +v2 payloads rather than interpreting their status under the wrong semantics. + ### Size cap The default 16 MiB cap can be overridden via the pytest CLI option or an ini setting: @@ -177,3 +181,14 @@ does not discard normal Results from that worker. same version everywhere. - `pytest-xdist` itself does not support interactive debugging (`--pdb`, `--trace`); use single-process mode for debugging. + +The private xdist envelope is versioned independently from public result data. +The v2 projection carries optional terminal evaluation, trace end reason, turn +evaluation purpose, and trial population provenance together. This contract +layer does not change verdict cadence, so the fields are additive within v2. +The first execution layer that switches to terminal-trace verdict semantics +must bump the envelope before mixed versions could combine different verdict +bases. Oversized-result markers retain population provenance when the marker +still fits its hard cap. Pathologically large provenance is omitted with an +explicit `_rampart_population_ref_omitted` marker rather than violating the +transport limit. diff --git a/rampart/attacks/_factory.py b/rampart/attacks/_factory.py index 33a796cd..bd3fca49 100644 --- a/rampart/attacks/_factory.py +++ b/rampart/attacks/_factory.py @@ -73,8 +73,8 @@ def xpia( Benign user request(s) that cause the agent to process poisoned content. evaluator (Evaluator): What condition to check for. - max_turns (int): Maximum prompt-response exchanges before - ERROR. Defaults to 5. + max_turns (int): Maximum prompt-response exchanges. Reaching the + limit resolves the trace normally. Defaults to 5. event_handlers (list[ExecutionEventHandler] | None): Optional additional handlers for custom observability. diff --git a/rampart/attacks/_xpia.py b/rampart/attacks/_xpia.py index 637e422f..2fc535df 100644 --- a/rampart/attacks/_xpia.py +++ b/rampart/attacks/_xpia.py @@ -71,8 +71,8 @@ class XPIAExecution(BaseExecution): attachments. driver (PromptDriver): How to drive the trigger conversation. evaluator (Evaluator): What condition to check for. - max_turns (int): Maximum prompt-response exchanges before the - execution stops with ERROR. Prevents unbounded loops. + max_turns (int): Maximum prompt-response exchanges. Reaching the + limit resolves the trace normally and prevents unbounded loops. event_handlers (list[ExecutionEventHandler] | None): Additional handlers beyond the framework defaults. """ diff --git a/rampart/core/__init__.py b/rampart/core/__init__.py index c4612a32..4ca46703 100644 --- a/rampart/core/__init__.py +++ b/rampart/core/__init__.py @@ -33,11 +33,20 @@ SafetyStatus, resolve_as_attack, resolve_as_probe, + resolve_attack_verdict, + resolve_probe_verdict, +) +from rampart.core.trace import ( + EvaluationRecord, + TraceRun, + evaluate_terminal_async, + run_trace_async, ) from rampart.core.types import ( EvalContext, EvalOutcome, EvalResult, + EvaluationPurpose, ObservabilityLevel, Payload, PayloadFormat, @@ -45,6 +54,7 @@ Response, SideEffect, ToolCall, + TraceEndReason, Turn, ) @@ -58,6 +68,8 @@ "EvalContext", "EvalOutcome", "EvalResult", + "EvaluationPurpose", + "EvaluationRecord", "Evaluator", "ExecutionEvent", "ExecutionEventData", @@ -86,9 +98,15 @@ "Surface", "ToolCall", "ToolDeclaration", + "TraceEndReason", + "TraceRun", "Turn", + "evaluate_terminal_async", "evaluate_turn_async", "execute_trials_async", "resolve_as_attack", "resolve_as_probe", + "resolve_attack_verdict", + "resolve_probe_verdict", + "run_trace_async", ] diff --git a/rampart/core/_population.py b/rampart/core/_population.py new file mode 100644 index 00000000..f8f81853 --- /dev/null +++ b/rampart/core/_population.py @@ -0,0 +1,124 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Shared validation for trial population configuration and provenance.""" + +from __future__ import annotations + +import math + + +def validate_population_id(value: object) -> str: + """Validate and return a population identifier. + + Returns: + str: Validated population identifier. + + Raises: + TypeError: If ``value`` is not a string. + ValueError: If ``value`` is empty or exceeds the transport bound. + """ + if not isinstance(value, str): + msg = "population id must be a string" + raise TypeError(msg) + if not value: + msg = "population id must be non-empty" + raise ValueError(msg) + return value + + +def validate_population_size( + value: object, + *, + name: str, + allow_zero: bool = False, +) -> int: + """Validate and return a population size. + + Returns: + int: Validated population size. + + Raises: + TypeError: If ``value`` is not a non-boolean integer. + ValueError: If ``value`` is outside the supported range. + """ + if type(value) is not int: + msg = f"{name} must be a non-boolean integer" + raise TypeError(msg) + minimum = 0 if allow_zero else 1 + if value < minimum: + if minimum == 0: + msg = f"{name} must be greater than or equal to 0" + else: + msg = f"{name} must be greater than or equal to 1" + raise ValueError(msg) + return value + + +def validate_population_threshold(value: object, *, name: str) -> float: + """Validate and return a finite population threshold. + + Returns: + float: Normalized population threshold. + + Raises: + TypeError: If ``value`` is not a non-boolean number. + ValueError: If ``value`` is non-finite or outside [0.0, 1.0]. + """ + if isinstance(value, bool) or not isinstance(value, int | float): + msg = f"{name} must be a number" + raise TypeError(msg) + try: + normalized = float(value) + except OverflowError as exc: + msg = f"{name} must be finite" + raise ValueError(msg) from exc + if not math.isfinite(normalized): + msg = f"{name} must be finite" + raise ValueError(msg) + if not 0.0 <= normalized <= 1.0: + msg = f"{name} must be between 0.0 and 1.0" + raise ValueError(msg) + return normalized + + +def validate_population_index(value: object, *, size: int) -> int: + """Validate and return a population member index. + + Returns: + int: Validated population index. + + Raises: + TypeError: If ``value`` is not a non-boolean integer. + ValueError: If ``value`` falls outside the population. + """ + if type(value) is not int: + msg = "population index must be an integer" + raise TypeError(msg) + if not 0 <= value < size: + msg = "population index must be between 0 and size - 1" + raise ValueError(msg) + return value + + +def validate_population_parameters( + *, + size: object, + threshold: object, + size_name: str, + threshold_name: str, + allow_empty: bool = False, +) -> tuple[int, float]: + """Validate and normalize shared population parameters. + + Returns: + tuple[int, float]: Validated size and normalized threshold. + """ + return ( + validate_population_size( + size, + name=size_name, + allow_zero=allow_empty, + ), + validate_population_threshold(threshold, name=threshold_name), + ) diff --git a/rampart/core/execution.py b/rampart/core/execution.py index 769bc43d..c00fba23 100644 --- a/rampart/core/execution.py +++ b/rampart/core/execution.py @@ -18,6 +18,7 @@ from enum import Enum from typing import TYPE_CHECKING, Protocol, runtime_checkable +from rampart.core._population import validate_population_parameters from rampart.core.result import PopulationRef, PopulationResult, Result, SafetyStatus from rampart.core.types import ( EvalContext, @@ -377,7 +378,7 @@ async def execute_trials_async( 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, threshold = _validate_trial_parameters( n=n, threshold=threshold, ) @@ -407,22 +408,22 @@ def _validate_trial_parameters( *, n: int, threshold: float, -) -> None: +) -> tuple[int, float]: """Validate trial population parameters. + Returns: + tuple[int, float]: Validated count and normalized threshold. + 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) + return validate_population_parameters( + size=n, + threshold=threshold, + size_name="n", + threshold_name="threshold", + ) async def evaluate_turn_async( diff --git a/rampart/core/result.py b/rampart/core/result.py index 2683a2e3..ac95d1e7 100644 --- a/rampart/core/result.py +++ b/rampart/core/result.py @@ -16,10 +16,16 @@ from typing import TYPE_CHECKING, Any from rampart.common.text import safe_str, safe_str_list +from rampart.core._population import ( + validate_population_id, + validate_population_index, + validate_population_parameters, +) from rampart.core.types import ( EvalOutcome, EvalResult, ObservabilityLevel, + TraceEndReason, Turn, ) @@ -98,9 +104,9 @@ class PopulationRef: """Identifies the trial population that a Result belongs to. Args: - id: Unique identifier shared by every result in the population. + id: Non-empty identifier shared by every result in the population. index: Zero-based position of the result within the population. - size: Number of results requested for the population. + size: Positive number of results requested for the population. threshold: Required safe-result rate for the population. """ @@ -109,6 +115,26 @@ class PopulationRef: size: int threshold: float + def __post_init__(self) -> None: + """Validate bounded, internally consistent population provenance. + + Raises: + TypeError: If a field has the wrong runtime type. + ValueError: If a field is empty or out of range. + """ + population_id = validate_population_id(self.id) + size, threshold = validate_population_parameters( + size=self.size, + threshold=self.threshold, + size_name="population size", + threshold_name="population threshold", + ) + index = validate_population_index(self.index, size=size) + object.__setattr__(self, "id", population_id) + object.__setattr__(self, "index", index) + object.__setattr__(self, "size", size) + object.__setattr__(self, "threshold", threshold) + @dataclass(kw_only=True) class Result: @@ -133,7 +159,14 @@ class Result: that a report states a level someone chose rather than one the framework assumed. Built-in strategies pass ``adapter.observability_profile``. + terminal_evaluation: Evaluator output for the terminal trace. It is an + input to status; execution policy may adjust the final status. + None for manual/error results and execution strategies that have + not migrated to terminal-trace verdicts. turns: The full conversation for evidence and debugging. + trace_end_reason: Why the trace stopped producing turns. None when + execution failed before normal termination or the producing + strategy has not migrated to trace-end provenance. duration_seconds: How long the test execution took. harm_category: Which harm category this test covers. Accepts HarmCategory enum values for built-in categories or plain strings @@ -149,7 +182,9 @@ class Result: status: SafetyStatus summary: str observability_level: ObservabilityLevel + terminal_evaluation: EvalResult | None = None turns: list[Turn] = field(default_factory=list[Turn]) + trace_end_reason: TraceEndReason | None = None duration_seconds: float = 0.0 harm_category: HarmCategory | str | None = None strategy: str = "" @@ -169,10 +204,19 @@ def safe(self) -> bool: return self.status is SafetyStatus.SAFE @property - def eval_results(self) -> list[EvalResult]: - """Evaluator outcomes derived from turns.""" + def turn_evaluations(self) -> list[EvalResult]: + """Online evaluator outcomes attached to turns.""" return [t.eval_result for t in self.turns if t.eval_result is not None] + @property + def eval_results(self) -> list[EvalResult]: + """Compatibility view of online evaluations attached to turns. + + ``terminal_evaluation`` is intentionally excluded. New consumers + should use ``turn_evaluations`` for online evidence. + """ + return self.turn_evaluations + def __bool__(self) -> bool: """Assert-safe: bool(result) means the agent behaved safely. @@ -221,9 +265,13 @@ def __post_init__(self) -> None: 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) + _, self.threshold = validate_population_parameters( + size=len(self.results), + threshold=self.threshold, + size_name="population size", + threshold_name="threshold", + allow_empty=True, + ) @property def safe_count(self) -> int: @@ -287,6 +335,66 @@ def __repr__(self) -> str: ) +def resolve_attack_verdict(*, evaluation: EvalResult) -> SafetyStatus: + """Map one final evaluation using attack polarity. + + Args: + evaluation: The terminal-trace evaluator result. + + Returns: + SafetyStatus: DETECTED maps to UNSAFE, NOT_DETECTED maps to SAFE, + and UNDETERMINED is preserved. + + Raises: + ValueError: If ``evaluation.outcome`` is not a known EvalOutcome. + """ + outcome = _require_eval_outcome(evaluation=evaluation) + if outcome is EvalOutcome.DETECTED: + return SafetyStatus.UNSAFE + if outcome is EvalOutcome.NOT_DETECTED: + return SafetyStatus.SAFE + return SafetyStatus.UNDETERMINED + + +def resolve_probe_verdict(*, evaluation: EvalResult) -> SafetyStatus: + """Map one final evaluation using probe polarity. + + Args: + evaluation: The terminal-trace evaluator result. + + Returns: + SafetyStatus: DETECTED maps to SAFE, NOT_DETECTED maps to UNSAFE, + and UNDETERMINED is preserved. + + Raises: + ValueError: If ``evaluation.outcome`` is not a known EvalOutcome. + """ + outcome = _require_eval_outcome(evaluation=evaluation) + if outcome is EvalOutcome.DETECTED: + return SafetyStatus.SAFE + if outcome is EvalOutcome.NOT_DETECTED: + return SafetyStatus.UNSAFE + return SafetyStatus.UNDETERMINED + + +def _require_eval_outcome(*, evaluation: EvalResult) -> EvalOutcome: + """Return a validated evaluator outcome. + + Raises: + ValueError: If a third-party evaluator returned an unknown value. + """ + if isinstance(evaluation.outcome, EvalOutcome): + return evaluation.outcome + msg = f"Unknown EvalOutcome: {evaluation.outcome!r}" + raise ValueError(msg) + + +def _validate_eval_results(*, eval_results: list[EvalResult]) -> None: + """Validate every outcome before applying legacy list precedence.""" + for evaluation in eval_results: + _require_eval_outcome(evaluation=evaluation) + + def resolve_as_attack(*, eval_results: list[EvalResult]) -> SafetyStatus: """Attack semantics: detected -> UNSAFE, not detected -> SAFE. @@ -307,6 +415,7 @@ def resolve_as_attack(*, eval_results: list[EvalResult]) -> SafetyStatus: """ if not eval_results: return SafetyStatus.ERROR + _validate_eval_results(eval_results=eval_results) if any(er.detected for er in eval_results): return SafetyStatus.UNSAFE if any(er.outcome == EvalOutcome.UNDETERMINED for er in eval_results): @@ -333,6 +442,7 @@ def resolve_as_probe(*, eval_results: list[EvalResult]) -> SafetyStatus: """ if not eval_results: return SafetyStatus.ERROR + _validate_eval_results(eval_results=eval_results) if any(er.outcome == EvalOutcome.NOT_DETECTED for er in eval_results): return SafetyStatus.UNSAFE if any(er.outcome == EvalOutcome.UNDETERMINED for er in eval_results): diff --git a/rampart/core/trace.py b/rampart/core/trace.py new file mode 100644 index 00000000..3b54ac9d --- /dev/null +++ b/rampart/core/trace.py @@ -0,0 +1,212 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Shared linear trace execution and terminal evaluation helpers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from typing import TYPE_CHECKING + +from rampart.core.types import ( + EvalContext, + EvalResult, + EvaluationPurpose, + ObservabilityLevel, + TraceEndReason, + Turn, +) + +if TYPE_CHECKING: + from rampart.core.adapter import Session + from rampart.core.evaluator import Evaluator + from rampart.core.manifest import AppManifest + from rampart.core.prompt_driver import PromptDriver + + +@dataclass(frozen=True, kw_only=True, eq=False) +class EvaluationRecord: + """One online evaluation and the exact context it judged. + + Args: + evaluator: Evaluator object that produced the result. Identity is the + reuse boundary. + context: Exact raw-trace context passed to the evaluator. + result: Evaluation returned for that context. + """ + + evaluator: Evaluator + context: EvalContext + result: EvalResult + + +@dataclass(kw_only=True) +class TraceRun: + """A completed linear trace and its latest online evaluation. + + ``turns`` is the driver/report view and may carry online evidence. + ``raw_turns`` is the evaluator view and never carries framework-produced + evaluation annotations. + + Args: + trace_end_reason: Why the trace stopped producing turns. + observability_level: What the adapter can observe. + manifest: Agent capabilities used to create evaluator contexts. + turns: Annotated history passed to prompt drivers and results. + raw_turns: Annotation-free history passed to evaluators. + latest_online_evaluation: Most recent stop-condition evaluation. + """ + + trace_end_reason: TraceEndReason + observability_level: ObservabilityLevel + manifest: AppManifest | None = None + turns: list[Turn] = field(default_factory=list[Turn]) + raw_turns: list[Turn] = field(default_factory=list[Turn]) + latest_online_evaluation: EvaluationRecord | None = None + + +def _evaluation_context( + *, + raw_turns: list[Turn], + observability_level: ObservabilityLevel, + manifest: AppManifest | None, +) -> EvalContext: + """Build an evaluator context from a snapshot of the raw trace. + + Returns: + EvalContext: Context holding a shallow snapshot of raw turns. + """ + return EvalContext( + turns=list(raw_turns), + observability_level=observability_level, + manifest=manifest, + ) + + +async def run_trace_async( + *, + session: Session, + driver: PromptDriver, + max_turns: int, + observability_level: ObservabilityLevel, + stop_when: Evaluator | None = None, + manifest: AppManifest | None = None, +) -> TraceRun: + """Drive a linear conversation with optional online stopping. + + The runner does not own session lifetime or exception conversion. Callers + keep the session context active around this function, and exceptions from + the driver, session, or evaluator propagate unchanged. + + Args: + session: Active agent session. + driver: Prompt source for the conversation. + max_turns: Maximum number of requests sent to the agent. + observability_level: What the adapter can observe. + stop_when: Optional evaluator checked after every response. A detected + outcome terminates the trace. + manifest: Agent capabilities exposed to evaluators. + + Returns: + TraceRun: Completed turns, termination reason, and online evidence. + + Raises: + ValueError: If ``max_turns`` is negative. + """ + if max_turns < 0: + msg = "max_turns must be non-negative." + raise ValueError(msg) + + run = TraceRun( + trace_end_reason=TraceEndReason.MAX_TURNS_REACHED, + observability_level=observability_level, + manifest=manifest, + ) + + for turn_index in range(max_turns): + decision = await driver.next_prompt_async(history=list(run.turns)) + if decision is None: + run.trace_end_reason = TraceEndReason.DRIVER_EXHAUSTED + return run + + response = await session.send_async(decision.request) + raw_turn = Turn( + request=decision.request, + response=response, + turn_number=turn_index, + driver_reasoning=decision.reasoning, + ) + run.raw_turns.append(raw_turn) + + if stop_when is None: + run.turns.append(raw_turn) + continue + + context = _evaluation_context( + raw_turns=run.raw_turns, + observability_level=observability_level, + manifest=manifest, + ) + evaluation = await stop_when.evaluate_async(context=context) + run.latest_online_evaluation = EvaluationRecord( + evaluator=stop_when, + context=context, + result=evaluation, + ) + run.turns.append( + replace( + raw_turn, + eval_result=evaluation, + eval_purpose=EvaluationPurpose.STOP_CHECK, + ), + ) + if evaluation.detected: + run.trace_end_reason = TraceEndReason.STOP_CONDITION_MET + return run + + return run + + +async def evaluate_terminal_async( + *, + evaluator: Evaluator, + run: TraceRun, +) -> EvalResult | None: + """Evaluate the terminal raw trace, reusing an identical online judgment. + + Args: + evaluator: Evaluator responsible for the final verdict. + run: Completed trace from :func:`run_trace_async`. + + Returns: + EvalResult | None: Final evaluation, or None when no turns exist. + + Call this before leaving any active session or injection context required + by the evaluator. Requests, responses, and their nested values are treated + as immutable after the runner appends them. + """ + if not run.raw_turns: + return None + + record = run.latest_online_evaluation + if ( + record is not None + and record.evaluator is evaluator + and len(record.context.turns) == len(run.raw_turns) + and all( + evaluated is terminal + for evaluated, terminal in zip( + record.context.turns, + run.raw_turns, + strict=True, + ) + ) + ): + return replace(record.result, evidence=list(record.result.evidence)) + + context = _evaluation_context( + raw_turns=run.raw_turns, + observability_level=run.observability_level, + manifest=run.manifest, + ) + return await evaluator.evaluate_async(context=context) diff --git a/rampart/core/types.py b/rampart/core/types.py index 96da246a..8bb39b1d 100644 --- a/rampart/core/types.py +++ b/rampart/core/types.py @@ -264,6 +264,31 @@ def __post_init__(self) -> None: raise ValueError(msg) +class EvaluationPurpose(Enum): + """Why an evaluation was attached to a turn. + + Attributes: + STOP_CHECK: The evaluation was produced by an online stop + condition. It is execution evidence, not the final verdict input. + """ + + STOP_CHECK = "stop_check" + + +class TraceEndReason(Enum): + """Why a trace stopped producing turns. + + Attributes: + DRIVER_EXHAUSTED: The prompt driver returned no next request. + MAX_TURNS_REACHED: The configured turn budget truncated the trace. + STOP_CONDITION_MET: An online stop condition fired. + """ + + DRIVER_EXHAUSTED = "driver_exhausted" + MAX_TURNS_REACHED = "max_turns_reached" + STOP_CONDITION_MET = "stop_condition_met" + + @dataclass(frozen=True, kw_only=True) class Turn: """One prompt-response exchange. @@ -276,6 +301,8 @@ class Turn: request: What was sent to the agent. response: What the agent returned. eval_result: Evaluator outcome for this turn. + eval_purpose: Why ``eval_result`` was produced. None when the purpose was + not recorded, including executions that predate the trace runner. turn_number: Position in the conversation (0-indexed). timestamp: When this exchange occurred. driver_reasoning: Why the driver chose this request. @@ -284,10 +311,21 @@ class Turn: request: Request response: Response eval_result: EvalResult | None = None + eval_purpose: EvaluationPurpose | None = None turn_number: int = 0 timestamp: datetime | None = None driver_reasoning: str = "" + def __post_init__(self) -> None: + """Validate evaluation annotation consistency. + + Raises: + ValueError: If an evaluation purpose is present without a result. + """ + if self.eval_purpose is not None and self.eval_result is None: + msg = "eval_purpose requires eval_result" + raise ValueError(msg) + class EvalOutcome(Enum): """What the evaluator determined. diff --git a/rampart/drivers/llm.py b/rampart/drivers/llm.py index 3416812f..1462a296 100644 --- a/rampart/drivers/llm.py +++ b/rampart/drivers/llm.py @@ -8,9 +8,9 @@ - The **driver-side conversation** with the driving LLM, stored in PyRIT's CentralMemory keyed by self._conversation_id. Each turn - consists of a framework-built user message (containing the latest - agent response and evaluator feedback) and the LLM's next-prompt - reply. + consists of a framework-built user message containing the latest agent + response and any available online evaluator feedback, followed by the + LLM's next-prompt reply. - The **agent-side conversation** with the agent under test, represented by the ``history: list[Turn]`` passed into @@ -73,10 +73,9 @@ class LLMDriver: represented by ``history: list[Turn]`` passed into ``next_prompt_async``. - Termination is handled externally: the evaluator's early-stop - (on detection) or the execution loop's max_turns budget. The - driver never self-terminates — empty LLM responses raise - ``DriverError`` rather than returning None. + Termination is handled externally by an explicit online stop condition or + the execution loop's max-turn budget. The driver never self-terminates — + empty LLM responses raise ``DriverError`` rather than returning None. One driver instance = one driver-side conversation. Construct a new driver per test. Use ``from_target`` for custom targets. diff --git a/rampart/probes/_factory.py b/rampart/probes/_factory.py index f0b109d0..4c216e84 100644 --- a/rampart/probes/_factory.py +++ b/rampart/probes/_factory.py @@ -25,6 +25,7 @@ def behavior( *, prompt: str, evaluator: Evaluator, + stop_when: Evaluator | None = None, max_turns: int = 25, event_handlers: list[ExecutionEventHandler] | None = None, ) -> BaseExecution: ... @@ -35,6 +36,7 @@ def behavior( *, prompts: list[str], evaluator: Evaluator, + stop_when: Evaluator | None = None, max_turns: int = 25, event_handlers: list[ExecutionEventHandler] | None = None, ) -> BaseExecution: ... @@ -45,6 +47,7 @@ def behavior( *, driver: PromptDriver, evaluator: Evaluator, + stop_when: Evaluator | None = None, max_turns: int = 25, event_handlers: list[ExecutionEventHandler] | None = None, ) -> BaseExecution: ... @@ -56,6 +59,7 @@ def behavior( prompts: list[str] | None = None, driver: PromptDriver | None = None, evaluator: Evaluator, + stop_when: Evaluator | None = None, max_turns: int = 25, event_handlers: list[ExecutionEventHandler] | None = None, ) -> BaseExecution: @@ -69,8 +73,10 @@ def behavior( prompts (list[str] | None): A list of prompt strings. driver (PromptDriver | None): A pre-built prompt driver. evaluator (Evaluator): What behavior to check for. - max_turns (int): Maximum prompt-response exchanges before - returning ERROR. Defaults to 25. + stop_when (Evaluator | None): Optional online condition that stops + the trace when detected. Defaults to None. + max_turns (int): Maximum prompt-response exchanges. Reaching the + limit resolves the trace normally. Defaults to 25. event_handlers (list[ExecutionEventHandler] | None): Optional additional handlers. @@ -95,6 +101,7 @@ def behavior( return SingleTurnExecution( driver=resolved_driver, evaluator=evaluator, + stop_when=stop_when, max_turns=max_turns, event_handlers=event_handlers, ) diff --git a/rampart/probes/_single_turn.py b/rampart/probes/_single_turn.py index 8a912500..f1370f54 100644 --- a/rampart/probes/_single_turn.py +++ b/rampart/probes/_single_turn.py @@ -1,12 +1,12 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. -"""SingleTurnExecution — probe execution strategy. +"""SingleTurnExecution — behavioral probe execution strategy. -Sends prompts via a PromptDriver, evaluates responses, and resolves -using probe semantics (DETECTED → SAFE, NOT_DETECTED → UNSAFE). -No injection phase — just session creation, prompt driving, evaluation, -and cleanup. Inherits BaseExecution lifecycle. +Sends prompts via a PromptDriver, evaluates the completed trace once, and +resolves using probe semantics (DETECTED → SAFE, NOT_DETECTED → UNSAFE). +No injection phase — just session creation, prompt driving, optional online +stopping, terminal evaluation, and cleanup. Inherits BaseExecution lifecycle. """ from __future__ import annotations @@ -14,32 +14,32 @@ import logging from typing import TYPE_CHECKING -from rampart.common.text import safe_str +from rampart.common.text import safe_str, safe_str_list from rampart.core.execution import ( BaseExecution, ExecutionEventHandler, - evaluate_turn_async, ) from rampart.core.result import ( Result, SafetyStatus, _explain_undetermined, _summarize_undetermined_operands, - resolve_as_probe, + resolve_probe_verdict, ) -from rampart.core.types import EvalOutcome +from rampart.core.trace import evaluate_terminal_async, run_trace_async +from rampart.core.types import TraceEndReason if TYPE_CHECKING: from rampart.core.adapter import AgentAdapter from rampart.core.evaluator import Evaluator from rampart.core.prompt_driver import PromptDriver - from rampart.core.types import EvalResult, Turn + from rampart.core.types import EvalResult logger = logging.getLogger(__name__) class SingleTurnExecution(BaseExecution): - """Executes a probe: send prompts, evaluate, resolve as probe. + """Execute a behavioral probe over a completed trace. Inherits BaseExecution. No injection phase — just session creation, prompt driving, evaluation, and cleanup. The lifecycle @@ -49,8 +49,10 @@ class SingleTurnExecution(BaseExecution): Args: driver (PromptDriver): How to drive the conversation. evaluator (Evaluator): What behavior to check for. - max_turns (int): Maximum prompt-response exchanges before - returning ERROR. Defaults to 25. + stop_when (Evaluator | None): Optional online condition that stops the + trace when detected. Defaults to None. + max_turns (int): Maximum prompt-response exchanges. Reaching the + limit resolves the trace normally. Defaults to 25. event_handlers (list[ExecutionEventHandler] | None): Additional handlers beyond the framework defaults. """ @@ -60,12 +62,14 @@ def __init__( *, driver: PromptDriver, evaluator: Evaluator, + stop_when: Evaluator | None = None, max_turns: int = 25, event_handlers: list[ExecutionEventHandler] | None = None, ) -> None: super().__init__(event_handlers=event_handlers) self._driver = driver self._evaluator = evaluator + self._stop_when = stop_when self._max_turns = max_turns @property @@ -82,38 +86,37 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result: Returns: Result: Safety verdict with full conversation evidence. """ - turns: list[Turn] = [] - async with await adapter.create_session_async() as session: - for turn_index in range(self._max_turns): - decision = await self._driver.next_prompt_async(history=turns) - if decision is None: - break - - response = await session.send_async(decision.request) - turn = await evaluate_turn_async( - evaluator=self._evaluator, - history=turns, - request=decision.request, - response=response, - turn_number=turn_index, - driver_reasoning=decision.reasoning, - manifest=adapter.manifest, - observability_level=adapter.observability_profile, - ) - turns.append(turn) - - if turn.eval_result and turn.eval_result.detected: - break - - eval_results = [t.eval_result for t in turns if t.eval_result is not None] - status = resolve_as_probe(eval_results=eval_results) + run = await run_trace_async( + session=session, + driver=self._driver, + max_turns=self._max_turns, + observability_level=adapter.observability_profile, + stop_when=self._stop_when, + manifest=adapter.manifest, + ) + evaluation = await evaluate_terminal_async( + evaluator=self._evaluator, + run=run, + ) + + status = ( + SafetyStatus.ERROR + if evaluation is None + else resolve_probe_verdict(evaluation=evaluation) + ) return Result( status=status, - summary=_build_summary(status=status, eval_results=eval_results), - turns=turns, - strategy="probe", + summary=_build_summary( + status=status, + evaluation=evaluation, + trace_end_reason=run.trace_end_reason, + ), + terminal_evaluation=evaluation, + turns=run.turns, + trace_end_reason=run.trace_end_reason, + strategy=self.strategy_name, observability_level=adapter.observability_profile, ) @@ -121,46 +124,54 @@ async def _execute_async(self, *, adapter: AgentAdapter) -> Result: def _build_summary( *, status: SafetyStatus, - eval_results: list[EvalResult], + evaluation: EvalResult | None, + trace_end_reason: TraceEndReason, ) -> str: """Build a human-readable one-line summary. Args: status (SafetyStatus): The resolved safety status. - eval_results (list[EvalResult]): The evaluator outputs. + evaluation (EvalResult | None): The terminal evaluator output. + trace_end_reason (TraceEndReason): Why the trace ended. Returns: str: A summary string for the Result. """ - if status == SafetyStatus.SAFE: - return "Expected behavior detected" + _summarize_undetermined_operands( + eval_results = [evaluation] if evaluation is not None else [] + budget_suffix = ( + " (turn budget exhausted)" + if trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + else "" + ) + if status is SafetyStatus.SAFE: + evidence = ( + safe_str_list(value=evaluation.evidence) if evaluation is not None else [] + ) + detail = "; ".join(evidence[:3]) + if not detail and evaluation is not None: + detail = safe_str(value=evaluation.rationale).strip() + suffix = f": {detail}" if detail else "" + gap_suffix = _summarize_undetermined_operands( eval_results=eval_results, ) - if status == SafetyStatus.UNSAFE: - # resolve_as_probe returns UNSAFE only when some evaluator was - # NOT_DETECTED, so the reason has to come from one of those. Taking any - # rationale would let an undetermined turn explain a definitive verdict. - # - # Rendered before the emptiness test, not after: a rationale whose - # truthiness raises would otherwise cost the verdict, and one that is - # only whitespace would render a summary with nothing after the colon. - rationales = [ - rendered - for er in eval_results - if er.outcome == EvalOutcome.NOT_DETECTED - and (rendered := safe_str(value=er.rationale).strip()) - ] - detail = rationales[-1] if rationales else "Expected behavior not detected" - return f"UNSAFE: {detail}" - if status == SafetyStatus.UNDETERMINED: + return f"Expected behavior detected{suffix}{gap_suffix}{budget_suffix}" + if status is SafetyStatus.UNSAFE: + detail = safe_str(value=evaluation.rationale).strip() if evaluation else "" + return f"UNSAFE: {detail or 'Expected behavior not detected'}{budget_suffix}" + if status is SafetyStatus.UNDETERMINED: detail = _explain_undetermined( eval_results=eval_results, fallback="Could not determine if expected behavior occurred", ) - return f"UNDETERMINED: {detail}" - detail = ( - safe_str(value=eval_results[-1].rationale) - if eval_results - else "No evaluation data" - ) - return f"ERROR: {detail}" + return f"UNDETERMINED: {detail}{budget_suffix}" + + detail = safe_str(value=evaluation.rationale).strip() if evaluation else "" + if detail: + error_detail = f"{detail}{budget_suffix}" + elif trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED: + error_detail = "Driver produced no turns" + elif trace_end_reason is TraceEndReason.MAX_TURNS_REACHED: + error_detail = "Turn budget exhausted before the agent was exercised" + else: + error_detail = "Probe produced no evaluable trace" + return f"ERROR: {error_detail}" diff --git a/rampart/pytest_plugin/_xdist.py b/rampart/pytest_plugin/_xdist.py index 3f321b00..07acb3d5 100644 --- a/rampart/pytest_plugin/_xdist.py +++ b/rampart/pytest_plugin/_xdist.py @@ -35,6 +35,7 @@ from rampart.core.types import ( EvalOutcome, EvalResult, + EvaluationPurpose, ObservabilityLevel, Payload, PayloadFormat, @@ -42,6 +43,7 @@ Response, SideEffect, ToolCall, + TraceEndReason, Turn, ) @@ -53,8 +55,8 @@ logger = logging.getLogger(__name__) -SCHEMA_VERSION: str = "rampart.xdist.v2" -WORKEROUTPUT_KEY: str = "rampart_xdist_v2" +SCHEMA_VERSION: str = "rampart.xdist.v3" +WORKEROUTPUT_KEY: str = "rampart_xdist_v3" REPORT_RESULTS_ATTR: str = "_rampart_results" SIZE_LIMIT_OPTION: str = "rampart_xdist_max_bytes" DEFAULT_SIZE_LIMIT_BYTES: int = 16 * 1024 * 1024 @@ -449,6 +451,9 @@ def _serialize_turn(*, turn: Turn, nodeid: str) -> dict[str, Any]: if turn.eval_result is not None else None ), + "eval_purpose": ( + turn.eval_purpose.value if turn.eval_purpose is not None else None + ), "turn_number": turn.turn_number, "timestamp": _isoformat(timestamp=turn.timestamp), "driver_reasoning": turn.driver_reasoning, @@ -467,12 +472,31 @@ def _serialize_injection_record(*, injection: InjectionRecord) -> dict[str, Any] } +def _serialize_population_ref( + *, + population: PopulationRef | None, +) -> dict[str, Any] | None: + """Serialize optional trial-population provenance. + + Returns: + dict[str, Any] | None: JSON-safe provenance, or None when absent. + """ + if population is None: + return None + return { + "id": population.id, + "index": population.index, + "size": population.size, + "threshold": population.threshold, + } + + def _serialize_result(*, result: Result, nodeid: str) -> dict[str, Any]: """Serialize a Result to a JSON-safe dict for the xdist transport. - This is the full-fidelity transport projection: it round-trips back - to a ``Result`` via :func:`_deserialize_result`, and intentionally - differs from the flatter public report shape produced by + This full-fidelity transport projection round-trips terminal and online + evaluation provenance together with trial-population attribution. It + intentionally differs from the flatter public report shape produced by ``JsonFileReportSink._serialize_result``. The two projections are deliberately separate (different fields, sanitization, and size handling) and must not be naively merged into one serializer. @@ -484,7 +508,17 @@ def _serialize_result(*, result: Result, nodeid: str) -> dict[str, Any]: "safe": result.safe, "status": result.status.value, "summary": result.summary, + "terminal_evaluation": ( + _serialize_eval_result(eval_result=result.terminal_evaluation) + if result.terminal_evaluation is not None + else None + ), "turns": [_serialize_turn(turn=t, nodeid=nodeid) for t in result.turns], + "trace_end_reason": ( + result.trace_end_reason.value + if result.trace_end_reason is not None + else None + ), "duration_seconds": safe_float(value=result.duration_seconds), "harm_category": ( str(result.harm_category) if result.harm_category is not None else None @@ -494,16 +528,7 @@ def _serialize_result(*, result: Result, nodeid: str) -> dict[str, Any]: "injections": [ _serialize_injection_record(injection=i) for i in result.injections ], - "population": ( - { - "id": result.population.id, - "index": result.population.index, - "size": result.population.size, - "threshold": result.population.threshold, - } - if result.population is not None - else None - ), + "population": _serialize_population_ref(population=result.population), "metadata": _sanitize_metadata( metadata=result.metadata, nodeid=nodeid, @@ -548,7 +573,7 @@ def _truncated_result_data( *, result: Result, nodeid: str, - size_bytes: int, + size_bytes: int | None, limit_bytes: int, ) -> dict[str, Any]: """Build a bounded ERROR Result marker for oversized transport data. @@ -573,7 +598,9 @@ def _truncated_result_data( "RAMPART Result exceeded the xdist transport size cap; " "full content was truncated." ), + "terminal_evaluation": None, "turns": [], + "trace_end_reason": None, "duration_seconds": 0.0, "harm_category": _bounded_attribution( value=harm_category, @@ -585,6 +612,7 @@ def _truncated_result_data( # part that overflowed. "observability_level": result.observability_level.value, "injections": [], + "population": None, "metadata": { "_pytest_test_name": _bounded_attribution( value=test_name, @@ -599,29 +627,43 @@ def _truncated_result_data( "_rampart_limit_bytes": limit_bytes, }, } - if _serialized_size(data=marker) <= limit_bytes: - return marker - logger.warning( - "Compacting truncation marker for %s to fit the %d-byte transport cap.", - _bounded_attribution( + marker_metadata = cast("dict[str, Any]", marker["metadata"]) + if _serialized_size(data=marker) > limit_bytes: + logger.warning( + "Compacting truncation marker for %s to fit the %d-byte transport cap.", + _bounded_attribution( + value=nodeid, + max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, + ), + limit_bytes, + ) + marker["harm_category"] = _bounded_attribution( + value=harm_category, + max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, + ) + marker_metadata["_pytest_test_name"] = _bounded_attribution( + value=test_name, + max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, + ) + marker_metadata["_pytest_nodeid"] = _bounded_attribution( value=nodeid, max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, - ), - limit_bytes, - ) - marker["harm_category"] = _bounded_attribution( - value=harm_category, - max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, - ) - marker_metadata = cast("dict[str, Any]", marker["metadata"]) - marker_metadata["_pytest_test_name"] = _bounded_attribution( - value=test_name, - max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, - ) - marker_metadata["_pytest_nodeid"] = _bounded_attribution( - value=nodeid, - max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, - ) + ) + + population = _serialize_population_ref(population=result.population) + if population is not None: + marker["population"] = population + try: + population_fits = _serialized_size(data=marker) <= limit_bytes + except (OverflowError, TypeError, ValueError): + population_fits = False + if not population_fits: + marker["population"] = None + marker_metadata["_rampart_population_ref_omitted"] = True + + if _serialized_size(data=marker) > limit_bytes: + marker["population"] = None + marker_metadata["_rampart_population_ref_omitted"] = True return marker @@ -657,22 +699,33 @@ def _serialize_capped_result( dict[str, Any]: Full Result data or a bounded truncation marker. """ data = _serialize_result(result=result, nodeid=nodeid) - size_bytes = _serialized_size(data=data) try: + size_bytes = _serialized_size(data=data) _enforce_result_size( size_bytes=size_bytes, limit_bytes=limit_bytes, nodeid=nodeid, ) + except (OverflowError, TypeError, ValueError) as exc: + logger.warning( + "Result for %r could not be serialized safely and was truncated: %s", + _bounded_attribution( + value=nodeid, + max_bytes=_TRUNCATED_FALLBACK_ATTRIBUTION_MAX_BYTES, + ), + safe_str(value=exc), + ) + size_bytes = None except SizeLimitError as exc: logger.warning("%s", exc) - return _truncated_result_data( - result=result, - nodeid=nodeid, - size_bytes=size_bytes, - limit_bytes=limit_bytes, - ) - return data + else: + return data + return _truncated_result_data( + result=result, + nodeid=nodeid, + size_bytes=size_bytes, + limit_bytes=limit_bytes, + ) def serialize_report_data( @@ -832,6 +885,48 @@ def _deserialize_eval_outcome(*, value: object) -> EvalOutcome: raise WorkerOutputError(msg) from exc +def _deserialize_evaluation_purpose(*, value: object) -> EvaluationPurpose | None: + """Deserialize an optional turn evaluation purpose. + + Returns: + EvaluationPurpose | None: The purpose, or None when absent. + + Raises: + WorkerOutputError: If ``value`` is not a known EvaluationPurpose. + """ + if value is None: + return None + if not isinstance(value, str): + msg = f"Expected string for EvaluationPurpose, got {type(value).__name__}." + raise WorkerOutputError(msg) + try: + return EvaluationPurpose(value) + except ValueError as exc: + msg = f"Unknown EvaluationPurpose value: {value!r}." + raise WorkerOutputError(msg) from exc + + +def _deserialize_trace_end_reason(*, value: object) -> TraceEndReason | None: + """Deserialize an optional trace end reason. + + Returns: + TraceEndReason | None: The reason, or None when absent. + + Raises: + WorkerOutputError: If ``value`` is not a known TraceEndReason. + """ + if value is None: + return None + if not isinstance(value, str): + msg = f"Expected string for TraceEndReason, got {type(value).__name__}." + raise WorkerOutputError(msg) + try: + return TraceEndReason(value) + except ValueError as exc: + msg = f"Unknown TraceEndReason value: {value!r}." + raise WorkerOutputError(msg) from exc + + def _deserialize_harm_category(*, value: object) -> HarmCategory | str | None: """Deserialize a HarmCategory enum value, plain string, or None. @@ -1116,10 +1211,18 @@ def _deserialize_turn(*, data: object) -> Turn: raise WorkerOutputError(msg) typed = cast("dict[str, Any]", data) raw_turn_number = typed.get("turn_number", 0) + eval_result = _deserialize_eval_result(data=typed.get("eval_result")) + eval_purpose = _deserialize_evaluation_purpose( + value=typed.get("eval_purpose"), + ) + if eval_purpose is not None and eval_result is None: + msg = "eval_purpose requires eval_result" + raise WorkerOutputError(msg) return Turn( request=_deserialize_request(data=typed.get("request")), response=_deserialize_response(data=typed.get("response")), - eval_result=_deserialize_eval_result(data=typed.get("eval_result")), + eval_result=eval_result, + eval_purpose=eval_purpose, turn_number=int(raw_turn_number) if isinstance(raw_turn_number, int) else 0, timestamp=_deserialize_datetime(value=typed.get("timestamp")), driver_reasoning=_strip_ansi(text=str(typed.get("driver_reasoning", ""))), @@ -1182,15 +1285,24 @@ def _deserialize_population_ref(*, data: object) -> PopulationRef | None: f"Expected number for population threshold, got {type(threshold).__name__}." ) raise WorkerOutputError(msg) - if not math.isfinite(threshold): + try: + normalized_threshold = float(threshold) + except (OverflowError, ValueError) as exc: + msg = f"Expected finite number for population threshold, got {threshold!r}." + raise WorkerOutputError(msg) from exc + if not math.isfinite(normalized_threshold): msg = f"Expected finite number for population threshold, got {threshold!r}." raise WorkerOutputError(msg) - return PopulationRef( - id=population_id, - index=index, - size=size, - threshold=float(threshold), - ) + try: + return PopulationRef( + id=population_id, + index=index, + size=size, + threshold=normalized_threshold, + ) + except (TypeError, ValueError) as exc: + msg = f"Invalid population provenance: {exc}" + raise WorkerOutputError(msg) from exc def _deserialize_result(*, data: object) -> Result: @@ -1215,19 +1327,31 @@ def _deserialize_result(*, data: object) -> Result: strip_ansi=True, ) raw_duration = typed.get("duration_seconds", 0.0) - duration = ( - float(raw_duration) - if isinstance(raw_duration, int | float) and math.isfinite(float(raw_duration)) - else 0.0 - ) + try: + duration = ( + float(raw_duration) + if isinstance(raw_duration, int | float) + and not isinstance(raw_duration, bool) + else 0.0 + ) + except (OverflowError, ValueError): + duration = 0.0 + if not math.isfinite(duration): + duration = 0.0 return Result( status=_deserialize_safety_status(value=typed.get("status")), summary=_strip_ansi(text=str(typed.get("summary", ""))), + terminal_evaluation=_deserialize_eval_result( + data=typed.get("terminal_evaluation"), + ), turns=[ _deserialize_turn(data=t) for t in cast("list[Any]", raw_turns if isinstance(raw_turns, list) else []) ], duration_seconds=duration, + trace_end_reason=_deserialize_trace_end_reason( + value=typed.get("trace_end_reason"), + ), harm_category=_deserialize_harm_category(value=typed.get("harm_category")), strategy=str(typed.get("strategy", "")), observability_level=_deserialize_observability_level( diff --git a/rampart/reporting/json_file.py b/rampart/reporting/json_file.py index 60ec4c3a..e04f3103 100644 --- a/rampart/reporting/json_file.py +++ b/rampart/reporting/json_file.py @@ -31,7 +31,7 @@ def pytest_rampart_sinks(config): from pathlib import Path from rampart.core.result import Result - from rampart.core.types import Turn + from rampart.core.types import EvalResult, Turn from rampart.reporting.sink import TestRunReport @@ -127,6 +127,16 @@ def _serialize_result(self, result: Result) -> dict[str, Any]: "safe": result.safe, "status": result.status.value, "summary": result.summary, + "terminal_evaluation": ( + self._serialize_eval_result(result.terminal_evaluation) + if result.terminal_evaluation is not None + else None + ), + "trace_end_reason": ( + result.trace_end_reason.value + if result.trace_end_reason is not None + else None + ), "harm_category": str(result.harm_category) if result.harm_category else None, @@ -181,6 +191,26 @@ def _serialize_turn(turn: Turn) -> dict[str, Any]: ) if operands: data["eval_undetermined_operands"] = operands + if turn.eval_purpose is not None: + data["eval_purpose"] = turn.eval_purpose.value if turn.driver_reasoning: data["driver_reasoning"] = turn.driver_reasoning return data + + @staticmethod + def _serialize_eval_result(eval_result: EvalResult) -> dict[str, Any]: + """Convert an EvalResult to the public report projection. + + Returns: + dict[str, Any]: JSON-serializable evaluator evidence. + """ + data: dict[str, Any] = { + "outcome": eval_result.outcome.value, + "confidence": safe_float(value=eval_result.confidence), + "evidence": safe_str_list(value=eval_result.evidence), + "rationale": safe_str(value=eval_result.rationale), + } + operands = safe_str_list(value=eval_result.undetermined_operands) + if operands: + data["undetermined_operands"] = operands + return data diff --git a/tests/unit/core/test_execution.py b/tests/unit/core/test_execution.py index 25008f2e..48527e29 100644 --- a/tests/unit/core/test_execution.py +++ b/tests/unit/core/test_execution.py @@ -4,6 +4,7 @@ import asyncio import types from typing import Self +from unittest.mock import MagicMock import pytest @@ -359,6 +360,23 @@ async def test_rejects_invalid_threshold_before_execution_async(self) -> None: assert handler.events == [] + @pytest.mark.parametrize("threshold", [True, float("nan"), float("inf")]) + async def test_rejects_malformed_threshold_before_factory_async( + self, + threshold: object, + ) -> None: + factory = MagicMock(return_value=_SuccessExecution()) + + with pytest.raises((TypeError, ValueError)): + await execute_trials_async( + execution_factory=factory, + adapter=_StubAdapter(), + n=1, + threshold=threshold, # ty: ignore[invalid-argument-type] + ) + + factory.assert_not_called() + class TestPopulationPublicExports: def test_execute_trials_exported_from_rampart(self) -> None: @@ -607,6 +625,7 @@ async def test_returns_turn_with_eval_result_async(self) -> None: assert turn.eval_result is not None assert turn.eval_result.outcome is EvalOutcome.DETECTED + assert turn.eval_purpose is None assert turn.request.prompt == "hello" assert turn.response.text == "world" assert turn.turn_number == 0 diff --git a/tests/unit/core/test_result.py b/tests/unit/core/test_result.py index 3f4dee99..6fe2e43b 100644 --- a/tests/unit/core/test_result.py +++ b/tests/unit/core/test_result.py @@ -6,11 +6,14 @@ Result, SafetyStatus, HarmCategory, resolve functions. """ +import warnings + import pytest from rampart.core.result import ( HarmCategory, InjectionRecord, + PopulationRef, PopulationResult, Result, SafetyStatus, @@ -18,6 +21,8 @@ _summarize_undetermined_operands, resolve_as_attack, resolve_as_probe, + resolve_attack_verdict, + resolve_probe_verdict, ) from rampart.core.types import ( EvalOutcome, @@ -25,6 +30,7 @@ ObservabilityLevel, Request, Response, + TraceEndReason, Turn, ) @@ -160,6 +166,20 @@ def test_defaults(self) -> None: assert r.observability_level is ObservabilityLevel.RESPONSE_ONLY assert r.injections == [] assert r.metadata == {} + assert r.terminal_evaluation is None + assert r.trace_end_reason is None + + def test_terminal_evaluation_and_trace_end_reason_round_trip(self) -> None: + evaluation = _er(EvalOutcome.DETECTED) + r = Result( + observability_level=ObservabilityLevel.RESPONSE_ONLY, + status=SafetyStatus.UNSAFE, + summary="bad", + terminal_evaluation=evaluation, + trace_end_reason=TraceEndReason.STOP_CONDITION_MET, + ) + assert r.terminal_evaluation is evaluation + assert r.trace_end_reason is TraceEndReason.STOP_CONDITION_MET def test_harm_category_accepts_enum(self) -> None: r = Result( @@ -260,6 +280,14 @@ def test_rejects_threshold_outside_valid_range(self, threshold: float) -> None: with pytest.raises(ValueError, match="threshold must be between"): PopulationResult(results=[], threshold=threshold) + @pytest.mark.parametrize("threshold", [True, float("nan"), float("inf")]) + def test_rejects_invalid_threshold(self, threshold: object) -> None: + with pytest.raises((TypeError, ValueError)): + PopulationResult( + results=[], + threshold=threshold, # ty: ignore[invalid-argument-type] + ) + def test_summary_contains_population_verdict(self) -> None: population = PopulationResult( results=[_result(SafetyStatus.SAFE), _result(SafetyStatus.UNSAFE)], @@ -282,8 +310,50 @@ def test_repr(self) -> None: ) -class TestResultEvalResultsProperty: - """eval_results is a property derived from turns.""" +class TestPopulationRef: + def test_accepts_generated_shape(self) -> None: + ref = PopulationRef(id="a" * 32, index=2, size=5, threshold=0.8) + assert ref.index == 2 + + @pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"id": ""}, "id must be non-empty"), + ({"index": -1}, "index must be"), + ({"index": 5}, "index must be"), + ({"size": 0}, "size must be"), + ({"threshold": float("nan")}, "threshold must be finite"), + ({"threshold": 1.1}, "threshold must be between"), + ], + ) + def test_rejects_invalid_provenance( + self, + overrides: dict[str, object], + message: str, + ) -> None: + values: dict[str, object] = { + "id": "population-1", + "index": 0, + "size": 5, + "threshold": 0.8, + } + values.update(overrides) + with pytest.raises((TypeError, ValueError), match=message): + PopulationRef(**values) + + def test_accepts_large_but_semantically_valid_provenance(self) -> None: + ref = PopulationRef( + id="x" * 513, + index=0, + size=2**31, + threshold=0.5, + ) + assert len(ref.id) == 513 + assert ref.size == 2**31 + + +class TestResultTurnEvaluationsProperty: + """Turn evaluations remain separate from the terminal evaluation.""" def test_empty_turns_gives_empty_eval_results(self) -> None: r = Result( @@ -291,6 +361,7 @@ def test_empty_turns_gives_empty_eval_results(self) -> None: status=SafetyStatus.SAFE, summary="ok", ) + assert r.turn_evaluations == [] assert r.eval_results == [] def test_turns_with_eval_results_returned_in_order(self) -> None: @@ -314,7 +385,8 @@ def test_turns_with_eval_results_returned_in_order(self) -> None: summary="bad", turns=turns, ) - assert r.eval_results == [er1, er2] + assert r.turn_evaluations == [er1, er2] + assert r.eval_results == r.turn_evaluations def test_turns_without_eval_result_filtered(self) -> None: er = _er(EvalOutcome.DETECTED) @@ -335,7 +407,26 @@ def test_turns_without_eval_result_filtered(self) -> None: summary="bad", turns=turns, ) - assert r.eval_results == [er] + assert r.turn_evaluations == [er] + + def test_final_evaluation_is_not_in_turn_eval_results(self) -> None: + final = _er(EvalOutcome.DETECTED) + turn_evaluation = _er(EvalOutcome.NOT_DETECTED) + r = Result( + observability_level=ObservabilityLevel.RESPONSE_ONLY, + status=SafetyStatus.UNSAFE, + summary="bad", + terminal_evaluation=final, + turns=[ + Turn( + request=Request(prompt="p"), + response=Response(text="r"), + eval_result=turn_evaluation, + ), + ], + ) + assert r.turn_evaluations == [turn_evaluation] + assert r.eval_results == [turn_evaluation] class TestResolveAsAttack: @@ -388,6 +479,13 @@ def test_all_not_detected_returns_safe(self) -> None: ) assert status is SafetyStatus.SAFE + def test_rejects_malformed_runtime_outcome(self) -> None: + malformed = EvalResult( + outcome="detected", # ty: ignore[invalid-argument-type] + ) + with pytest.raises(ValueError, match="Unknown EvalOutcome"): + resolve_as_attack(eval_results=[malformed]) + class TestResolveAsProbe: def test_empty_returns_error(self) -> None: @@ -439,6 +537,13 @@ def test_all_detected_returns_safe(self) -> None: ) assert status is SafetyStatus.SAFE + def test_rejects_malformed_runtime_outcome(self) -> None: + malformed = EvalResult( + outcome="detected", # ty: ignore[invalid-argument-type] + ) + with pytest.raises(ValueError, match="Unknown EvalOutcome"): + resolve_as_probe(eval_results=[malformed]) + class TestSummarizeUndeterminedOperands: def test_empty_when_nothing_was_undetermined(self) -> None: @@ -710,3 +815,59 @@ def test_ignores_blank_reasons(self) -> None: ) assert detail == "nothing to say" + + +def test_legacy_resolvers_remain_warning_free() -> None: + """The additive API does not start the legacy deprecation clock.""" + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert resolve_as_attack(eval_results=[]) is SafetyStatus.ERROR + assert resolve_as_probe(eval_results=[]) is SafetyStatus.ERROR + + +class TestResolveAttackVerdict: + @pytest.mark.parametrize( + ("evaluation", "expected"), + [ + (_er(EvalOutcome.DETECTED), SafetyStatus.UNSAFE), + (_er(EvalOutcome.NOT_DETECTED), SafetyStatus.SAFE), + (_er(EvalOutcome.UNDETERMINED), SafetyStatus.UNDETERMINED), + ], + ) + def test_maps_single_evaluation( + self, + evaluation: EvalResult, + expected: SafetyStatus, + ) -> None: + assert resolve_attack_verdict(evaluation=evaluation) is expected + + def test_rejects_malformed_runtime_outcome(self) -> None: + evaluation = EvalResult( + outcome="detected", # ty: ignore[invalid-argument-type] + ) + with pytest.raises(ValueError, match="Unknown EvalOutcome"): + resolve_attack_verdict(evaluation=evaluation) + + +class TestResolveProbeVerdict: + @pytest.mark.parametrize( + ("evaluation", "expected"), + [ + (_er(EvalOutcome.DETECTED), SafetyStatus.SAFE), + (_er(EvalOutcome.NOT_DETECTED), SafetyStatus.UNSAFE), + (_er(EvalOutcome.UNDETERMINED), SafetyStatus.UNDETERMINED), + ], + ) + def test_maps_single_evaluation( + self, + evaluation: EvalResult, + expected: SafetyStatus, + ) -> None: + assert resolve_probe_verdict(evaluation=evaluation) is expected + + def test_rejects_malformed_runtime_outcome(self) -> None: + evaluation = EvalResult( + outcome="detected", # ty: ignore[invalid-argument-type] + ) + with pytest.raises(ValueError, match="Unknown EvalOutcome"): + resolve_probe_verdict(evaluation=evaluation) diff --git a/tests/unit/core/test_trace.py b/tests/unit/core/test_trace.py new file mode 100644 index 00000000..924099de --- /dev/null +++ b/tests/unit/core/test_trace.py @@ -0,0 +1,301 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for the shared linear trace runner.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from rampart.core.evaluator import Evaluator +from rampart.core.manifest import AppManifest +from rampart.core.prompt_driver import PromptDecision +from rampart.core.trace import evaluate_terminal_async, run_trace_async +from rampart.core.types import ( + EvalOutcome, + EvalResult, + EvaluationPurpose, + ObservabilityLevel, + Request, + Response, + TraceEndReason, + Turn, +) +from rampart.drivers.static import StaticDriver +from tests.fixtures import MockSession + + +def _session(*responses: str) -> MockSession: + """Build a session returning the supplied response texts.""" + return MockSession(responses=[Response(text=text) for text in responses]) + + +def _evaluator(*outcomes: EvalOutcome) -> AsyncMock: + """Build an evaluator mock returning outcomes in order.""" + evaluator = AsyncMock(spec=Evaluator) + evaluator.evaluate_async.side_effect = [ + EvalResult(outcome=outcome, rationale=f"call {index}") + for index, outcome in enumerate(outcomes) + ] + return evaluator + + +class TestRunTraceAsync: + async def test_driver_exhaustion_returns_raw_turns_async(self) -> None: + run = await run_trace_async( + session=_session("r1", "r2"), + driver=StaticDriver(prompts=["p1", "p2"]), + max_turns=3, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) + + assert run.trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED + assert [turn.response.text for turn in run.turns] == ["r1", "r2"] + assert run.turns == run.raw_turns + assert run.latest_online_evaluation is None + + async def test_turn_budget_is_a_normal_termination_async(self) -> None: + run = await run_trace_async( + session=_session("r1", "r2", "r3"), + driver=StaticDriver(prompts=["p1", "p2", "p3"]), + max_turns=2, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) + + assert run.trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + assert len(run.turns) == 2 + + async def test_zero_budget_does_not_call_driver_async(self) -> None: + driver = AsyncMock() + + run = await run_trace_async( + session=_session("unused"), + driver=driver, + max_turns=0, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) + + assert run.trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + assert run.turns == [] + driver.next_prompt_async.assert_not_awaited() + + async def test_stop_condition_annotates_only_public_history_async(self) -> None: + evaluator = _evaluator(EvalOutcome.NOT_DETECTED, EvalOutcome.DETECTED) + manifest = AppManifest(name="agent") + + run = await run_trace_async( + session=_session("r1", "r2", "r3"), + driver=StaticDriver(prompts=["p1", "p2", "p3"]), + max_turns=3, + observability_level=ObservabilityLevel.TOOL_ONLY, + stop_when=evaluator, + manifest=manifest, + ) + + assert run.trace_end_reason is TraceEndReason.STOP_CONDITION_MET + assert len(run.turns) == 2 + assert all( + turn.eval_purpose is EvaluationPurpose.STOP_CHECK for turn in run.turns + ) + assert all(turn.eval_result is not None for turn in run.turns) + assert all(turn.eval_result is None for turn in run.raw_turns) + contexts = [ + call.kwargs["context"] for call in evaluator.evaluate_async.await_args_list + ] + assert [len(context.turns) for context in contexts] == [1, 2] + assert all( + turn.eval_result is None for context in contexts for turn in context.turns + ) + assert all( + context.observability_level is ObservabilityLevel.TOOL_ONLY + for context in contexts + ) + assert contexts[-1].manifest is manifest + + async def test_driver_cannot_mutate_owned_history_list_async(self) -> None: + class MutatingDriver: + def __init__(self) -> None: + self.calls = 0 + + async def next_prompt_async( + self, + *, + history: list[Turn], + ) -> PromptDecision | None: + history.append( + Turn( + request=Request(prompt="injected"), + response=Response(text="injected"), + ), + ) + if self.calls: + return None + self.calls += 1 + return PromptDecision(request=Request(prompt="p")) + + run = await run_trace_async( + session=_session("r"), + driver=MutatingDriver(), + max_turns=2, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) + + assert len(run.turns) == 1 + assert run.turns[0].request.prompt == "p" + + async def test_evaluator_exception_propagates_async(self) -> None: + evaluator = AsyncMock(spec=Evaluator) + evaluator.evaluate_async.side_effect = RuntimeError("judge failed") + + with pytest.raises(RuntimeError, match="judge failed"): + await run_trace_async( + session=_session("r"), + driver=StaticDriver(prompts=["p"]), + max_turns=1, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + stop_when=evaluator, + ) + + +class TestEvaluateTerminalAsync: + async def test_empty_trace_skips_evaluator_async(self) -> None: + evaluator = _evaluator(EvalOutcome.DETECTED) + run = await run_trace_async( + session=_session("unused"), + driver=StaticDriver(prompts=[]), + max_turns=1, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) + + result = await evaluate_terminal_async(evaluator=evaluator, run=run) + + assert result is None + assert run.trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED + evaluator.evaluate_async.assert_not_awaited() + + @pytest.mark.parametrize( + "outcomes", + [ + (EvalOutcome.DETECTED,), + (EvalOutcome.NOT_DETECTED,), + ], + ) + async def test_reuses_identical_latest_online_evaluation_async( + self, + outcomes: tuple[EvalOutcome, ...], + ) -> None: + evaluator = _evaluator(*outcomes) + run = await run_trace_async( + session=_session("r"), + driver=StaticDriver(prompts=["p"]), + max_turns=1, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + stop_when=evaluator, + ) + online_result = run.latest_online_evaluation + assert online_result is not None + + result = await evaluate_terminal_async(evaluator=evaluator, run=run) + + assert result == online_result.result + assert result is not online_result.result + assert result.evidence is not online_result.result.evidence + assert evaluator.evaluate_async.await_count == 1 + + async def test_non_firing_stop_reuses_terminal_prefix_without_extra_call_async( + self, + ) -> None: + evaluator = _evaluator( + EvalOutcome.NOT_DETECTED, + EvalOutcome.NOT_DETECTED, + EvalOutcome.NOT_DETECTED, + ) + run = await run_trace_async( + session=_session("r1", "r2", "r3"), + driver=StaticDriver(prompts=["p1", "p2", "p3"]), + max_turns=3, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + stop_when=evaluator, + ) + + result = await evaluate_terminal_async(evaluator=evaluator, run=run) + + assert result is not None and result.outcome is EvalOutcome.NOT_DETECTED + assert evaluator.evaluate_async.await_count == 3 + + async def test_distinct_evaluator_runs_once_on_terminal_trace_async(self) -> None: + stop = _evaluator(EvalOutcome.NOT_DETECTED) + verdict = _evaluator(EvalOutcome.DETECTED) + manifest = AppManifest(name="agent") + run = await run_trace_async( + session=_session("r"), + driver=StaticDriver(prompts=["p"]), + max_turns=1, + observability_level=ObservabilityLevel.RESPONSE_ONLY, + stop_when=stop, + manifest=manifest, + ) + + result = await evaluate_terminal_async( + evaluator=verdict, + run=run, + ) + + assert result is not None and result.outcome is EvalOutcome.DETECTED + verdict.evaluate_async.assert_awaited_once() + context = verdict.evaluate_async.await_args.kwargs["context"] + assert context.turns == run.raw_turns + assert context.observability_level is ObservabilityLevel.RESPONSE_ONLY + assert all(turn.eval_result is None for turn in context.turns) + + async def test_post_run_trace_mutation_prevents_reuse_async(self) -> None: + evaluator = _evaluator(EvalOutcome.NOT_DETECTED, EvalOutcome.DETECTED) + run = await run_trace_async( + session=_session("r", "later"), + driver=StaticDriver(prompts=["p"]), + max_turns=1, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + stop_when=evaluator, + ) + run.raw_turns.append( + Turn( + request=Request(prompt="later"), + response=Response(text="later"), + turn_number=1, + ), + ) + + result = await evaluate_terminal_async(evaluator=evaluator, run=run) + + assert result is not None and result.outcome is EvalOutcome.DETECTED + assert evaluator.evaluate_async.await_count == 2 + + async def test_undetermined_stop_does_not_terminate_async(self) -> None: + evaluator = _evaluator( + EvalOutcome.UNDETERMINED, + EvalOutcome.NOT_DETECTED, + ) + run = await run_trace_async( + session=_session("r1", "r2"), + driver=StaticDriver(prompts=["p1", "p2"]), + max_turns=2, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + stop_when=evaluator, + ) + + assert run.trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + assert len(run.turns) == 2 + assert run.turns[0].eval_purpose is EvaluationPurpose.STOP_CHECK + + +async def test_negative_turn_budget_raises_async() -> None: + """Negative budgets are rejected rather than treated as zero.""" + with pytest.raises(ValueError, match="non-negative"): + await run_trace_async( + session=_session("unused"), + driver=StaticDriver(prompts=[]), + max_turns=-1, + observability_level=ObservabilityLevel.TOOL_AND_SIDE_EFFECTS, + ) diff --git a/tests/unit/core/test_types.py b/tests/unit/core/test_types.py index 52b63655..7671f859 100644 --- a/tests/unit/core/test_types.py +++ b/tests/unit/core/test_types.py @@ -12,6 +12,7 @@ EvalContext, EvalOutcome, EvalResult, + EvaluationPurpose, ObservabilityLevel, Payload, PayloadFormat, @@ -19,6 +20,7 @@ Response, SideEffect, ToolCall, + TraceEndReason, Turn, ) @@ -112,6 +114,7 @@ def test_construction_with_defaults(self): assert t.timestamp is None assert t.driver_reasoning == "" assert t.eval_result is None + assert t.eval_purpose is None def test_eval_result_round_trips(self): er = EvalResult(outcome=EvalOutcome.DETECTED, rationale="found it") @@ -123,6 +126,24 @@ def test_eval_result_round_trips(self): assert t.eval_result is er assert t.eval_result is not None and t.eval_result.detected is True + def test_eval_purpose_round_trips(self): + evaluation = EvalResult(outcome=EvalOutcome.DETECTED) + t = Turn( + request=Request(prompt="p"), + response=Response(text="r"), + eval_result=evaluation, + eval_purpose=EvaluationPurpose.STOP_CHECK, + ) + assert t.eval_purpose is EvaluationPurpose.STOP_CHECK + + def test_eval_purpose_requires_eval_result(self) -> None: + with pytest.raises(ValueError, match="eval_purpose requires eval_result"): + Turn( + request=Request(prompt="p"), + response=Response(text="r"), + eval_purpose=EvaluationPurpose.STOP_CHECK, + ) + def test_frozen_prevents_mutation(self): t = Turn(request=Request(prompt="p"), response=Response(text="r")) with pytest.raises(dataclasses.FrozenInstanceError): @@ -150,6 +171,42 @@ def test_defaults(self): assert er.undetermined_operands == [] +class TestExecutionMetadataEnums: + def test_evaluation_purpose_value(self) -> None: + assert EvaluationPurpose.STOP_CHECK.value == "stop_check" + assert not isinstance(EvaluationPurpose.STOP_CHECK, str) + + def test_trace_end_reason_values(self) -> None: + assert TraceEndReason.DRIVER_EXHAUSTED.value == "driver_exhausted" + assert TraceEndReason.MAX_TURNS_REACHED.value == "max_turns_reached" + assert TraceEndReason.STOP_CONDITION_MET.value == "stop_condition_met" + assert not isinstance(TraceEndReason.STOP_CONDITION_MET, str) + + def test_purpose_and_reason_do_not_compare_equal(self) -> None: + assert EvaluationPurpose.STOP_CHECK != TraceEndReason.STOP_CONDITION_MET + + +def test_new_contract_is_available_from_core_package() -> None: + """Execution vocabulary is available from the narrower core API.""" + from rampart.core import ( + EvaluationPurpose as CoreEvaluationPurpose, + ) + from rampart.core import ( + TraceEndReason as CoreTraceEndReason, + ) + from rampart.core import ( + resolve_attack_verdict as core_attack_resolver, + ) + from rampart.core import ( + resolve_probe_verdict as core_probe_resolver, + ) + + assert CoreEvaluationPurpose is EvaluationPurpose + assert CoreTraceEndReason is TraceEndReason + assert core_attack_resolver is not None + assert core_probe_resolver is not None + + class TestEvalContext: def _make_turn( self, diff --git a/tests/unit/probes/test_single_turn.py b/tests/unit/probes/test_single_turn.py index f04b903b..f363845b 100644 --- a/tests/unit/probes/test_single_turn.py +++ b/tests/unit/probes/test_single_turn.py @@ -5,6 +5,9 @@ from __future__ import annotations +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock + import pytest from rampart.core.errors import InfrastructureError @@ -21,11 +24,13 @@ Request, Response, ToolCall, + TraceEndReason, Turn, ) from rampart.drivers.static import StaticDriver from rampart.evaluators import ( ResponseContains, + ResponseScope, SideEffectOccurred, ToolCalled, ) @@ -33,6 +38,10 @@ from rampart.probes._single_turn import _build_summary from tests.fixtures import MockAdapter, MockSession +if TYPE_CHECKING: + import types + from typing import Self + class _Unrenderable: """Stands in for an evaluator value whose ``__str__`` raises.""" @@ -111,6 +120,54 @@ async def next_prompt_async( return PromptDecision(request=Request(prompt="fresh")) +class _ToolSequence(BaseEvaluator): + """Detect a tool sequence across the complete trace.""" + + def __init__(self, *tool_names: str) -> None: + self._tool_names = tool_names + + async def evaluate_async(self, *, context: EvalContext) -> EvalResult: + observed = [call.name for call in context.all_tool_calls] + next_index = 0 + for name in observed: + if ( + next_index < len(self._tool_names) + and name == self._tool_names[next_index] + ): + next_index += 1 + return EvalResult( + outcome=( + EvalOutcome.DETECTED + if next_index == len(self._tool_names) + else EvalOutcome.NOT_DETECTED + ), + rationale=f"Observed tools: {observed}", + ) + + +class _RequiresPrior(BaseEvaluator): + """Detect when an action is always preceded by a required tool.""" + + def __init__(self, *, action: str, required: str) -> None: + self._action = action + self._required = required + + async def evaluate_async(self, *, context: EvalContext) -> EvalResult: + seen_required = False + for call in context.all_tool_calls: + if call.name == self._required: + seen_required = True + if call.name == self._action and not seen_required: + return EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale=f"{self._action} occurred before {self._required}", + ) + return EvalResult( + outcome=EvalOutcome.DETECTED, + rationale=f"{self._required} preceded {self._action}", + ) + + class TestProbePolarity: """Probe polarity: DETECTED -> SAFE, NOT_DETECTED -> UNSAFE.""" @@ -366,6 +423,8 @@ async def create_session_async(self): assert result.safe is False assert result.status == SafetyStatus.ERROR assert "InfrastructureError" in result.summary + assert result.terminal_evaluation is None + assert result.trace_end_reason is None class TestProbeEndToEnd: @@ -423,7 +482,7 @@ async def test_assert_pattern_async(self) -> None: class TestProbeMaxTurns: - """Max turns resolves normally via resolve_as_probe.""" + """Max turns resolves normally from the terminal evaluation.""" async def test_max_turns_resolves_normally_async(self) -> None: adapter = _adapter(responses=[Response(text="ok")]) @@ -437,170 +496,113 @@ async def test_max_turns_resolves_normally_async(self) -> None: assert result.safe is False assert result.status == SafetyStatus.UNSAFE assert len(result.turns) == 2 + assert result.trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + assert "turn budget exhausted" in result.summary -class TestProbeUnsafeSummary: - """An unsafe summary names the turn that settled it, not an undetermined one.""" - - def test_summary_uses_only_not_detected_rationales(self) -> None: - summary = _build_summary( - status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale="Target pattern not found in response text", - ), - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale="Adapter observability is 'tool_only'", - ), - ], - ) - - assert "not found" in summary - assert "tool_only" not in summary +class TestProbeSummary: + """Terminal evaluation summaries preserve evidence and observability gaps.""" - def test_summary_falls_back_without_a_rationale(self) -> None: + def test_unsafe_summary_uses_terminal_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.UNSAFE, - eval_results=[EvalResult(outcome=EvalOutcome.NOT_DETECTED)], + evaluation=EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale="Target pattern not found in response text", + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) - assert summary == "UNSAFE: Expected behavior not detected" + assert summary == "UNSAFE: Target pattern not found in response text" - def test_summary_falls_back_past_a_whitespace_rationale(self) -> None: + @pytest.mark.parametrize("rationale", ["", " "]) + def test_unsafe_summary_falls_back_without_a_rationale( + self, + rationale: str, + ) -> None: summary = _build_summary( status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult(outcome=EvalOutcome.NOT_DETECTED, rationale=" "), - ], + evaluation=EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale=rationale, + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNSAFE: Expected behavior not detected" - def test_summary_names_the_last_undetected_turn(self) -> None: - summary = _build_summary( - status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale="Disclaimer not found on the first prompt", - ), - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale="Disclaimer not found on the retry", - ), - ], - ) - - assert summary == "UNSAFE: Disclaimer not found on the retry" - - -class TestProbeUndeterminedSummary: - """An undetermined summary should name every gap that was carried up.""" - - def test_summary_names_every_operand_gap(self) -> None: + def test_undetermined_summary_names_every_operand_gap(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale="Left operand undetermined: tool calls unobservable", - undetermined_operands=[ - "tool calls unobservable", - "side effects unobservable", - ], - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + undetermined_operands=[ + "tool calls unobservable", + "side effects unobservable", + ], + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) - assert "tool calls unobservable" in summary - assert "side effects unobservable" in summary + assert summary == ( + "UNDETERMINED: tool calls unobservable; side effects unobservable" + ) - def test_summary_deduplicates_operand_reasons(self) -> None: + def test_undetermined_summary_deduplicates_operand_reasons(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - undetermined_operands=["same gap"], - ), - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - undetermined_operands=["same gap"], - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + undetermined_operands=["same gap", "same gap"], + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNDETERMINED: same gap" - def test_summary_counts_the_gaps_it_does_not_name(self) -> None: + def test_undetermined_summary_counts_extra_gaps(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - undetermined_operands=["gap a", "gap b", "gap c", "gap d"], - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + undetermined_operands=["gap a", "gap b", "gap c", "gap d"], + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNDETERMINED: gap a; gap b (and 2 more)" - def test_summary_ignores_operands_carried_by_a_settled_result(self) -> None: - summary = _build_summary( - status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - undetermined_operands=["gap that did not settle the verdict"], - ), - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale="Adapter observability is 'tool_only'", - ), - ], - ) - - assert "tool_only" in summary - assert "did not settle" not in summary - - def test_summary_falls_back_to_the_rationale(self) -> None: + def test_undetermined_summary_falls_back_to_the_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale="Adapter observability is 'response_only'", - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + rationale="Adapter observability is 'response_only'", + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNDETERMINED: Adapter observability is 'response_only'" - def test_summary_falls_back_without_a_rationale(self) -> None: + def test_undetermined_summary_falls_back_without_a_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[EvalResult(outcome=EvalOutcome.UNDETERMINED)], + evaluation=EvalResult(outcome=EvalOutcome.UNDETERMINED), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == ( "UNDETERMINED: Could not determine if expected behavior occurred" ) - -class TestProbeSafeSummary: - """A safe probe says when part of the evaluation stayed undetermined.""" - def test_safe_summary_names_the_undetermined_operand(self) -> None: summary = _build_summary( status=SafetyStatus.SAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.DETECTED, - undetermined_operands=["tool calls not reported"], - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.DETECTED, + undetermined_operands=["tool calls not reported"], + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == ( @@ -611,7 +613,8 @@ def test_safe_summary_names_the_undetermined_operand(self) -> None: def test_safe_summary_is_plain_when_everything_was_determined(self) -> None: summary = _build_summary( status=SafetyStatus.SAFE, - eval_results=[EvalResult(outcome=EvalOutcome.DETECTED)], + evaluation=EvalResult(outcome=EvalOutcome.DETECTED), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "Expected behavior detected" @@ -660,12 +663,11 @@ class TestProbeSummaryHostileOperands: def test_safe_summary_survives_a_bad_operand_collection(self) -> None: summary = _build_summary( status=SafetyStatus.SAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.DETECTED, - undetermined_operands=123, # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.DETECTED, + undetermined_operands=123, # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "Expected behavior detected" @@ -673,13 +675,12 @@ def test_safe_summary_survives_a_bad_operand_collection(self) -> None: def test_undetermined_summary_falls_back_past_a_bad_collection(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale="Adapter observability is 'tool_only'", - undetermined_operands=123, # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + rationale="Adapter observability is 'tool_only'", + undetermined_operands=123, # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNDETERMINED: Adapter observability is 'tool_only'" @@ -691,12 +692,11 @@ class TestProbeSummaryHostileRationale: def test_unsafe_summary_survives_a_raising_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNSAFE: " @@ -704,12 +704,11 @@ def test_unsafe_summary_survives_a_raising_rationale(self) -> None: def test_error_summary_survives_a_raising_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.ERROR, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "ERROR: " @@ -728,12 +727,11 @@ def strip(self, chars: str | None = None) -> str: summary = _build_summary( status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale=Sneaky(" the disclaimer was missing "), - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale=Sneaky(" the disclaimer was missing "), + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNSAFE: the disclaimer was missing" @@ -741,12 +739,11 @@ def strip(self, chars: str | None = None) -> str: def test_undetermined_summary_survives_a_raising_rationale(self) -> None: summary = _build_summary( status=SafetyStatus.UNDETERMINED, - eval_results=[ - EvalResult( - outcome=EvalOutcome.UNDETERMINED, - rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.UNDETERMINED, + rationale=_Unrenderable(), # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNDETERMINED: " @@ -761,12 +758,316 @@ def __str__(self) -> str: summary = _build_summary( status=SafetyStatus.UNSAFE, - eval_results=[ - EvalResult( - outcome=EvalOutcome.NOT_DETECTED, - rationale=RaisingBool(), # ty: ignore[invalid-argument-type] - ), - ], + evaluation=EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale=RaisingBool(), # ty: ignore[invalid-argument-type] + ), + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, ) assert summary == "UNSAFE: unrenderable rationale" + + +class TestProbeFinalTraceCadence: + async def test_verdict_evaluator_runs_once_over_complete_trace_async(self) -> None: + evaluator = AsyncMock() + evaluator.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.DETECTED, + ) + adapter = _adapter( + responses=[Response(text="r1"), Response(text="r2"), Response(text="r3")], + ) + + result = await Probes.behavior( + prompts=["p1", "p2", "p3"], + evaluator=evaluator, + ).execute_async(adapter=adapter) + + evaluator.evaluate_async.assert_awaited_once() + context = evaluator.evaluate_async.await_args.kwargs["context"] + assert len(context.turns) == 3 + assert result.terminal_evaluation is evaluator.evaluate_async.return_value + assert result.eval_results == [] + assert result.trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED + + async def test_tool_sequence_resolves_from_complete_trace_async(self) -> None: + result = await Probes.behavior( + prompts=["first", "second"], + evaluator=_ToolSequence("a", "b"), + ).execute_async( + adapter=_adapter( + responses=[ + Response(text="one", tool_calls=[ToolCall(name="a")]), + Response(text="two", tool_calls=[ToolCall(name="b")]), + ], + ), + ) + + assert result.status is SafetyStatus.SAFE + assert len(result.turns) == 2 + + async def test_requires_prior_observes_action_before_resolving_async( + self, + ) -> None: + result = await Probes.behavior( + prompts=["confirm", "delete"], + evaluator=_RequiresPrior(action="delete", required="confirm"), + ).execute_async( + adapter=_adapter( + responses=[ + Response(text="confirmed", tool_calls=[ToolCall(name="confirm")]), + Response(text="deleted", tool_calls=[ToolCall(name="delete")]), + ], + ), + ) + + assert result.status is SafetyStatus.SAFE + assert len(result.turns) == 2 + + async def test_zero_turns_returns_error_without_evaluation_async(self) -> None: + evaluator = AsyncMock() + result = await Probes.behavior( + prompts=[], + evaluator=evaluator, + ).execute_async(adapter=_adapter(responses=[Response(text="unused")])) + + assert result.status is SafetyStatus.ERROR + assert result.terminal_evaluation is None + assert result.trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED + evaluator.evaluate_async.assert_not_awaited() + + async def test_zero_turn_budget_returns_error_with_budget_reason_async( + self, + ) -> None: + evaluator = AsyncMock() + result = await Probes.behavior( + prompts=["unused"], + evaluator=evaluator, + max_turns=0, + ).execute_async(adapter=_adapter(responses=[Response(text="unused")])) + + assert result.status is SafetyStatus.ERROR + assert result.trace_end_reason is TraceEndReason.MAX_TURNS_REACHED + assert "budget" in result.summary.lower() + evaluator.evaluate_async.assert_not_awaited() + + async def test_default_driver_history_has_no_evaluator_feedback_async( + self, + ) -> None: + class RecordingDriver: + def __init__(self) -> None: + self.histories: list[list[Turn]] = [] + + async def next_prompt_async( + self, + *, + history: list[Turn], + ) -> PromptDecision | None: + self.histories.append(history) + if len(history) >= 2: + return None + return PromptDecision(request=Request(prompt=f"p{len(history)}")) + + driver = RecordingDriver() + result = await Probes.behavior( + driver=driver, + evaluator=_DetectsAlways(), + ).execute_async( + adapter=_adapter(responses=[Response(text="r1"), Response(text="r2")]), + ) + + assert len(result.turns) == 2 + assert all( + turn.eval_result is None for history in driver.histories for turn in history + ) + + async def test_explicit_identical_stop_reuses_fired_evaluation_async(self) -> None: + evaluator = AsyncMock() + evaluator.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.DETECTED, + rationale="stop now", + ) + + result = await Probes.behavior( + prompts=["p1", "p2"], + evaluator=evaluator, + stop_when=evaluator, + ).execute_async(adapter=_adapter(responses=[Response(text="r1")])) + + assert len(result.turns) == 1 + assert result.status is SafetyStatus.SAFE + assert result.trace_end_reason is TraceEndReason.STOP_CONDITION_MET + assert evaluator.evaluate_async.await_count == 1 + + async def test_distinct_stop_and_verdict_evaluators_do_not_cross_reuse_async( + self, + ) -> None: + stop = AsyncMock() + stop.evaluate_async.side_effect = [ + EvalResult(outcome=EvalOutcome.NOT_DETECTED), + EvalResult(outcome=EvalOutcome.DETECTED), + ] + verdict = AsyncMock() + verdict.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.DETECTED, + rationale="terminal verdict", + ) + + result = await Probes.behavior( + prompts=["p1", "p2", "p3"], + evaluator=verdict, + stop_when=stop, + ).execute_async( + adapter=_adapter(responses=[Response(text="r1"), Response(text="r2")]), + ) + + assert len(result.turns) == 2 + assert stop.evaluate_async.await_count == 2 + verdict.evaluate_async.assert_awaited_once() + context = verdict.evaluate_async.await_args.kwargs["context"] + assert len(context.turns) == 2 + assert result.terminal_evaluation is verdict.evaluate_async.return_value + + async def test_explicit_stop_feedback_is_available_to_driver_async(self) -> None: + class RecordingDriver: + def __init__(self) -> None: + self.histories: list[list[Turn]] = [] + + async def next_prompt_async( + self, + *, + history: list[Turn], + ) -> PromptDecision | None: + self.histories.append(history) + if len(history) >= 2: + return None + return PromptDecision(request=Request(prompt=f"p{len(history)}")) + + stop = AsyncMock() + stop.evaluate_async.side_effect = [ + EvalResult(outcome=EvalOutcome.NOT_DETECTED, rationale="continue"), + EvalResult(outcome=EvalOutcome.DETECTED, rationale="stop"), + ] + driver = RecordingDriver() + + await Probes.behavior( + driver=driver, + evaluator=stop, + stop_when=stop, + ).execute_async( + adapter=_adapter(responses=[Response(text="r1"), Response(text="r2")]), + ) + + second_history = driver.histories[1] + assert second_history[0].eval_result is not None + assert second_history[0].eval_result.rationale == "continue" + + async def test_all_turns_scope_applies_to_complete_probe_trace_async(self) -> None: + result = await Probes.behavior( + prompts=["p1", "p2"], + evaluator=ResponseContains("ok", scope=ResponseScope.ALL_TURNS), + ).execute_async( + adapter=_adapter(responses=[Response(text="no"), Response(text="ok")]), + ) + + assert len(result.turns) == 2 + assert result.status is SafetyStatus.UNSAFE + + async def test_negated_any_turn_scope_applies_to_complete_probe_trace_async( + self, + ) -> None: + result = await Probes.behavior( + prompts=["p1", "p2"], + evaluator=~ResponseContains( + "forbidden", + scope=ResponseScope.ANY_TURN, + ), + ).execute_async( + adapter=_adapter(responses=[Response(text="clean"), Response(text="safe")]), + ) + + assert result.status is SafetyStatus.SAFE + + async def test_unspecified_scope_warns_through_probe_execution_async(self) -> None: + with pytest.warns(FutureWarning, match="ResponseScope"): + result = await Probes.behavior( + prompts=["p1", "p2"], + evaluator=ResponseContains("ok"), + ).execute_async( + adapter=_adapter( + responses=[Response(text="not yet"), Response(text="ok")], + ), + ) + + assert result.status is SafetyStatus.SAFE + + async def test_terminal_evaluation_runs_before_session_close_async(self) -> None: + class RecordingSession: + def __init__(self) -> None: + self.closed = False + + async def send_async(self, request: Request) -> Response: + return Response(text=request.prompt or "") + + async def __aenter__(self) -> Self: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: types.TracebackType | None, + ) -> None: + self.closed = True + + session = RecordingSession() + + class Adapter: + manifest = AppManifest(name="test-agent") + observability_profile = ObservabilityLevel.RESPONSE_ONLY + + async def create_session_async(self): + return session + + class CheckingEvaluator(BaseEvaluator): + async def evaluate_async(self, *, context: EvalContext) -> EvalResult: + assert session.closed is False + return EvalResult(outcome=EvalOutcome.DETECTED) + + result = await Probes.behavior( + prompt="hello", + evaluator=CheckingEvaluator(), + ).execute_async(adapter=Adapter()) + + assert result.status is SafetyStatus.SAFE + assert session.closed is True + + async def test_safe_summary_includes_terminal_evidence_async(self) -> None: + evaluator = AsyncMock() + evaluator.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.DETECTED, + evidence=["terminal evidence"], + rationale="terminal rationale", + ) + + result = await Probes.behavior( + prompt="p", + evaluator=evaluator, + ).execute_async(adapter=_adapter(responses=[Response(text="r")])) + + assert "terminal evidence" in result.summary + + async def test_undetermined_summary_includes_terminal_rationale_async(self) -> None: + evaluator = AsyncMock() + evaluator.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.UNDETERMINED, + rationale="not enough evidence", + ) + + result = await Probes.behavior( + prompt="p", + evaluator=evaluator, + ).execute_async(adapter=_adapter(responses=[Response(text="r")])) + + assert result.status is SafetyStatus.UNDETERMINED + assert "not enough evidence" in result.summary diff --git a/tests/unit/pytest_plugin/test_xdist.py b/tests/unit/pytest_plugin/test_xdist.py index 416eea91..131f8d99 100644 --- a/tests/unit/pytest_plugin/test_xdist.py +++ b/tests/unit/pytest_plugin/test_xdist.py @@ -14,6 +14,8 @@ import pytest +from rampart.core.execution import BaseExecution, execute_trials_async +from rampart.core.manifest import AppManifest from rampart.core.result import ( HarmCategory, InjectionRecord, @@ -24,12 +26,14 @@ from rampart.core.types import ( EvalOutcome, EvalResult, + EvaluationPurpose, ObservabilityLevel, PayloadFormat, Request, Response, SideEffect, ToolCall, + TraceEndReason, Turn, ) from rampart.pytest_plugin._session import RampartSession @@ -60,6 +64,7 @@ serialize_worker_data, ) from rampart.reporting.sink import TestRunReport +from tests.fixtures import MockAdapter def _make_result( @@ -94,6 +99,7 @@ def _make_turn( prompt: str = "hi", text: str = "ok", eval_result: EvalResult | None = None, + eval_purpose: EvaluationPurpose | None = None, turn_number: int = 0, timestamp: datetime | None = None, driver_reasoning: str = "", @@ -102,6 +108,7 @@ def _make_turn( request=Request(prompt=prompt), response=Response(text=text), eval_result=eval_result, + eval_purpose=eval_purpose, turn_number=turn_number, timestamp=timestamp, driver_reasoning=driver_reasoning, @@ -493,6 +500,86 @@ def test_a_non_numeric_confidence_is_not_read_as_full(self) -> None: class TestResultFieldSerializationRoundTrip: + def test_terminal_contract_and_population_round_trip_together(self) -> None: + population = PopulationRef( + id="p" * 513, + index=2, + size=2**31, + threshold=0.8, + ) + terminal = _make_eval_result( + outcome=EvalOutcome.DETECTED, + evidence=["terminal evidence"], + ) + turn = _make_turn( + eval_result=_make_eval_result(), + eval_purpose=EvaluationPurpose.STOP_CHECK, + ) + result = _make_result(turns=[turn], population=population) + result.terminal_evaluation = terminal + result.trace_end_reason = TraceEndReason.STOP_CONDITION_MET + payload = _serialize_session_results( + session=_make_session_with_results(results_by_nodeid={"n": [result]}), + ) + + recovered = _deserialize_report_results(data=payload)["n"][0] + + assert recovered.population == population + assert recovered.terminal_evaluation is not None + assert recovered.terminal_evaluation.evidence == ["terminal evidence"] + assert recovered.trace_end_reason is TraceEndReason.STOP_CONDITION_MET + assert recovered.turns[0].eval_purpose is EvaluationPurpose.STOP_CHECK + + async def test_execute_trials_terminal_provenance_round_trip_async(self) -> None: + terminal_evaluation = _make_eval_result( + outcome=EvalOutcome.NOT_DETECTED, + evidence=["terminal evidence"], + ) + + class TerminalExecution(BaseExecution): + @property + def strategy_name(self) -> str: + return "terminal-test" + + async def _execute_async(self, *, adapter) -> Result: + del adapter + return Result( + status=SafetyStatus.SAFE, + summary="safe terminal trace", + observability_level=ObservabilityLevel.RESPONSE_ONLY, + terminal_evaluation=terminal_evaluation, + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, + ) + + adapter = MockAdapter( + responses=[Response(text="unused")], + manifest=AppManifest(name="agent"), + ) + population = await execute_trials_async( + execution_factory=TerminalExecution, + adapter=adapter, + n=2, + threshold=0.5, + ) + payload = serialize_report_data( + config=_make_config(is_worker=True), + nodeid="n", + results=population.results, + ) + + recovered = _deserialize_report_results(data=payload)["n"] + + assert len(recovered) == 2 + population_ids = { + result.population.id for result in recovered if result.population + } + assert len(population_ids) == 1 + assert all(result.terminal_evaluation is not None for result in recovered) + assert all( + result.trace_end_reason is TraceEndReason.DRIVER_EXHAUSTED + for result in recovered + ) + def test_datetime_round_trip(self) -> None: when = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC) turn = _make_turn(timestamp=when) @@ -586,15 +673,89 @@ def test_rejects_unknown_schema_version(self) -> None: with pytest.raises(SchemaVersionError, match="does not match"): deserialize_report_data(data=payload, report_nodeid="n") - def test_rejects_legacy_schema_version(self) -> None: + def test_rejects_previous_schema_version(self) -> None: payload: dict[str, Any] = { - "schema": "rampart.xdist.v1", + "schema": "rampart.xdist.v2", "nodeid": "n", "results": [], } with pytest.raises(SchemaVersionError, match="does not match"): deserialize_report_data(data=payload, report_nodeid="n") + @pytest.mark.parametrize( + ("field", "value"), + [ + ("trace_end_reason", "future_reason"), + ("eval_purpose", "future_purpose"), + ("trace_end_reason", ["driver_exhausted"]), + ("eval_purpose", {"purpose": "stop_check"}), + ], + ) + def test_rejects_unknown_terminal_contract_enum( + self, + field: str, + value: object, + ) -> None: + turn: dict[str, Any] = { + "request": {"prompt": "p"}, + "response": {"text": "r"}, + } + result_data: dict[str, Any] = { + "status": "safe", + "summary": "x", + "observability_level": "response_only", + "turns": [turn], + } + (turn if field == "eval_purpose" else result_data)[field] = value + payload = { + "schema": SCHEMA_VERSION, + "nodeid": "n", + "results": [result_data], + } + + with pytest.raises(WorkerOutputError, match=r"Unknown|Expected string"): + deserialize_report_data(data=payload, report_nodeid="n") + + def test_rejects_eval_purpose_without_eval_result(self) -> None: + payload = { + "schema": SCHEMA_VERSION, + "nodeid": "n", + "results": [ + { + "status": "safe", + "summary": "x", + "observability_level": "response_only", + "turns": [ + { + "request": {"prompt": "p"}, + "response": {"text": "r"}, + "eval_purpose": "stop_check", + }, + ], + }, + ], + } + + with pytest.raises(WorkerOutputError, match="eval_purpose requires"): + deserialize_report_data(data=payload, report_nodeid="n") + + def test_huge_duration_is_sanitized_to_zero(self) -> None: + payload = { + "schema": SCHEMA_VERSION, + "nodeid": "n", + "results": [ + { + "status": "safe", + "summary": "x", + "observability_level": "response_only", + "duration_seconds": 10**10_000, + }, + ], + } + + recovered = _deserialize_report_results(data=payload)["n"][0] + assert recovered.duration_seconds == pytest.approx(0.0) + def test_rejects_nodeid_mismatch(self) -> None: payload = {"schema": SCHEMA_VERSION, "nodeid": "other", "results": []} with pytest.raises(WorkerOutputError, match="does not match"): @@ -979,8 +1140,8 @@ def test_records_incomplete_on_legacy_workeroutput_key(self) -> None: node = MagicMock() node.gateway.id = "gw1" node.workeroutput = { - "rampart_xdist_v1": { - "schema": "rampart.xdist.v1", + "rampart_xdist_v2": { + "schema": "rampart.xdist.v2", "streamed_result_count": 0, }, } @@ -1240,6 +1401,86 @@ def test_oversized_result_is_localized_and_marks_incomplete( == 1 ) + def test_oversized_result_preserves_population_provenance(self) -> None: + population = PopulationRef( + id="population-1", + index=3, + size=5, + threshold=0.8, + ) + payload = serialize_report_data( + config=_make_config(is_worker=True, max_bytes=1024), + nodeid="n", + results=[ + _make_result( + summary="x" * 10_000, + population=population, + ), + ], + ) + recovered, truncated = deserialize_report_data( + data=payload, + report_nodeid="n", + ) + + assert truncated is True + assert recovered["n"][0].population == population + marker = payload["results"][0] + assert len(json.dumps(marker).encode("utf-8")) <= MIN_RESULT_SIZE_LIMIT_BYTES + + def test_oversized_population_provenance_is_omitted_from_marker(self) -> None: + population = PopulationRef( + id="\U0001f600" * 10_000, + index=0, + size=1, + threshold=1.0, + ) + payload = serialize_report_data( + config=_make_config(is_worker=True, max_bytes=1024), + nodeid="n", + results=[ + _make_result( + summary="x" * 10_000, + population=population, + ), + ], + ) + + recovered, truncated = deserialize_report_data( + data=payload, + report_nodeid="n", + ) + marker = payload["results"][0] + + assert truncated is True + assert recovered["n"][0].population is None + assert marker["metadata"]["_rampart_population_ref_omitted"] is True + assert len(json.dumps(marker).encode("utf-8")) <= MIN_RESULT_SIZE_LIMIT_BYTES + + def test_unserializable_integer_provenance_becomes_bounded_marker(self) -> None: + population = PopulationRef( + id="population-1", + index=0, + size=1 << 20_000, + threshold=1.0, + ) + payload = serialize_report_data( + config=_make_config(is_worker=True, max_bytes=1024), + nodeid="n", + results=[_make_result(population=population)], + ) + + recovered, truncated = deserialize_report_data( + data=payload, + report_nodeid="n", + ) + marker = payload["results"][0] + + assert truncated is True + assert recovered["n"][0].population is None + assert marker["metadata"]["_rampart_population_ref_omitted"] is True + assert len(json.dumps(marker).encode("utf-8")) <= MIN_RESULT_SIZE_LIMIT_BYTES + @pytest.mark.parametrize( "escaped", ["\x00" * 10_000, "\U0001f600" * 10_000], @@ -1300,11 +1541,11 @@ class TestConstants: def test_default_size_limit_is_16mb(self) -> None: assert DEFAULT_SIZE_LIMIT_BYTES == 16 * 1024 * 1024 - def test_schema_version_is_v2(self) -> None: - assert SCHEMA_VERSION == "rampart.xdist.v2" + def test_schema_version_is_v3(self) -> None: + assert SCHEMA_VERSION == "rampart.xdist.v3" def test_workeroutput_key_namespaced(self) -> None: - assert WORKEROUTPUT_KEY == "rampart_xdist_v2" + assert WORKEROUTPUT_KEY == "rampart_xdist_v3" class TestTestRunReportTestable: diff --git a/tests/unit/pytest_plugin/test_xdist_aggregation.py b/tests/unit/pytest_plugin/test_xdist_aggregation.py index 4d2933f4..c6042e28 100644 --- a/tests/unit/pytest_plugin/test_xdist_aggregation.py +++ b/tests/unit/pytest_plugin/test_xdist_aggregation.py @@ -595,6 +595,100 @@ def test_trial_mixed_load(trial_config): assert report["passed"] == 3 assert report["failed"] == 1 + def test_execute_trials_terminal_provenance_crosses_worker_boundary( + self, + configured_pytester: Pytester, + ) -> None: + """Worker-produced trial results retain provenance in controller JSON.""" + configured_pytester.makepyfile( + test_terminal_population=""" + import pytest + + from rampart.core import BaseExecution, execute_trials_async + from rampart.core.result import Result, SafetyStatus + from rampart.core.types import ( + EvalOutcome, + EvalResult, + EvaluationPurpose, + ObservabilityLevel, + Request, + Response, + TraceEndReason, + Turn, + ) + + class Adapter: + manifest = None + observability_profile = ObservabilityLevel.RESPONSE_ONLY + + class Execution(BaseExecution): + @property + def strategy_name(self): + return "terminal-population" + + async def _execute_async(self, *, adapter): + del adapter + online = EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + rationale="continue", + ) + terminal = EvalResult( + outcome=EvalOutcome.NOT_DETECTED, + evidence=["terminal evidence"], + rationale="complete", + ) + return Result( + status=SafetyStatus.SAFE, + summary="safe terminal trace", + observability_level=ObservabilityLevel.RESPONSE_ONLY, + terminal_evaluation=terminal, + trace_end_reason=TraceEndReason.DRIVER_EXHAUSTED, + turns=[Turn( + request=Request(prompt="p"), + response=Response(text="r"), + eval_result=online, + eval_purpose=EvaluationPurpose.STOP_CHECK, + )], + ) + + @pytest.mark.harm("test") + async def test_terminal_population_async(): + population = await execute_trials_async( + execution_factory=Execution, + adapter=Adapter(), + n=2, + threshold=0.5, + ) + assert population.safe + """, + ) + + result = configured_pytester.runpytest( + "-p", + "no:cacheprovider", + "-n", + "1", + ) + + result.assert_outcomes(passed=1) + reports = _load_reports(configured_pytester) + assert len(reports) == 1 + streamed = _report_results(reports[0]) + assert len(streamed) == 2 + populations = [item["population"] for item in streamed] + assert len({item["id"] for item in populations}) == 1 + assert [item["index"] for item in populations] == [0, 1] + assert all(item["size"] == 2 for item in populations) + assert all(item["threshold"] == pytest.approx(0.5) for item in populations) + assert all( + item["terminal_evaluation"]["evidence"] == ["terminal evidence"] + for item in streamed + ) + assert all(item["trace_end_reason"] == "driver_exhausted" for item in streamed) + assert all( + item["turns"][0]["eval_purpose"] == "stop_check" for item in streamed + ) + class TestXdistMetadata: def test_report_includes_xdist_metadata( diff --git a/tests/unit/reporting/test_json_file.py b/tests/unit/reporting/test_json_file.py index 6e32bfd9..bb394d79 100644 --- a/tests/unit/reporting/test_json_file.py +++ b/tests/unit/reporting/test_json_file.py @@ -5,6 +5,7 @@ from __future__ import annotations +import dataclasses import json from datetime import UTC, datetime from pathlib import Path @@ -18,11 +19,13 @@ from rampart.core.types import ( EvalOutcome, EvalResult, + EvaluationPurpose, ObservabilityLevel, Request, Response, SideEffect, ToolCall, + TraceEndReason, Turn, ) from rampart.reporting.json_file import JsonFileReportSink @@ -93,6 +96,34 @@ def test_population_is_null_for_single_execution(self) -> None: assert data["population"] is None + def test_terminal_contract_appears_with_population(self) -> None: + sink = JsonFileReportSink(output_dir=Path("/tmp")) + result = _result_with_turns() + result.population = PopulationRef( + id="population-1", + index=2, + size=5, + threshold=0.8, + ) + result.terminal_evaluation = EvalResult( + outcome=EvalOutcome.DETECTED, + evidence=["terminal evidence"], + rationale="terminal rationale", + ) + result.trace_end_reason = TraceEndReason.STOP_CONDITION_MET + result.turns[0] = dataclasses.replace( + result.turns[0], + eval_result=EvalResult(outcome=EvalOutcome.NOT_DETECTED), + eval_purpose=EvaluationPurpose.STOP_CHECK, + ) + + data = sink._serialize_result(result) + + assert data["population"]["id"] == "population-1" + assert data["terminal_evaluation"]["outcome"] == "detected" + assert data["trace_end_reason"] == "stop_condition_met" + assert data["turns"][0]["eval_purpose"] == "stop_check" + def test_result_reports_the_observability_level(self) -> None: # Not the value _result_with_turns defaults to, so a hardcoded # literal in the sink cannot satisfy this.