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/probes/behavioral.md b/docs/probes/behavioral.md index 1e13eba1..10709a60 100644 --- a/docs/probes/behavioral.md +++ b/docs/probes/behavioral.md @@ -92,7 +92,7 @@ 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`. | +| `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/results-and-reporting.md b/docs/usage/results-and-reporting.md index f1ff5856..abde89af 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,31 @@ 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. + +This layer makes terminal provenance durable before changing execution +cadence. Existing prefix-evaluated strategies leave these fields as `None` +until their follow-up migration; 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..3a297c1e 100644 --- a/docs/usage/xdist.md +++ b/docs/usage/xdist.md @@ -177,3 +177,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/probes/_factory.py b/rampart/probes/_factory.py index f0b109d0..271b14ec 100644 --- a/rampart/probes/_factory.py +++ b/rampart/probes/_factory.py @@ -69,8 +69,8 @@ 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. + 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. diff --git a/rampart/probes/_single_turn.py b/rampart/probes/_single_turn.py index 8a912500..eb010b11 100644 --- a/rampart/probes/_single_turn.py +++ b/rampart/probes/_single_turn.py @@ -49,8 +49,8 @@ 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. + 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. """ diff --git a/rampart/pytest_plugin/_xdist.py b/rampart/pytest_plugin/_xdist.py index 3f321b00..8145bc2e 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, ) @@ -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. @@ -888,13 +983,20 @@ def _deserialize_confidence(*, typed: dict[str, Any]) -> float: Returns: float: The reconstructed confidence, or ``NaN`` when it was present but not a usable finite number. + + Raises: + WorkerOutputError: If a numeric value cannot be converted to float. """ if "confidence" not in typed: return 1.0 raw_confidence = typed["confidence"] if isinstance(raw_confidence, bool) or not isinstance(raw_confidence, int | float): return math.nan - number = float(raw_confidence) + try: + number = float(raw_confidence) + except (OverflowError, ValueError) as exc: + msg = f"Confidence could not be converted to float: {raw_confidence!r}." + raise WorkerOutputError(msg) from exc return number if math.isfinite(number) else math.nan @@ -1116,10 +1218,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 +1292,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 +1334,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/pytest_plugin/test_plugin.py b/tests/unit/pytest_plugin/test_plugin.py index 2682cf18..34e19950 100644 --- a/tests/unit/pytest_plugin/test_plugin.py +++ b/tests/unit/pytest_plugin/test_plugin.py @@ -21,7 +21,11 @@ deactivate_collector, ) from rampart.pytest_plugin._session import RampartSession -from rampart.pytest_plugin._xdist import REPORT_RESULTS_ATTR, serialize_report_data +from rampart.pytest_plugin._xdist import ( + REPORT_RESULTS_ATTR, + SCHEMA_VERSION, + serialize_report_data, +) from rampart.pytest_plugin.plugin import ( _call_results_key, _emit_sinks, @@ -863,3 +867,28 @@ def test_malformed_envelope_marks_run_incomplete(self) -> None: ) pytest_runtest_logreport(cast("pytest.TestReport", report)) assert rampart_session.is_incomplete is True + + def test_overflowing_terminal_confidence_marks_run_incomplete(self) -> None: + nodeid = "test_plugin.py::test_stream" + report, rampart_session = _make_controller_report( + payload={ + "schema": SCHEMA_VERSION, + "nodeid": nodeid, + "results": [ + { + "status": "unsafe", + "summary": "unsafe terminal trace", + "observability_level": "response_only", + "terminal_evaluation": { + "outcome": "detected", + "confidence": 10**400, + }, + }, + ], + }, + ) + + pytest_runtest_logreport(cast("pytest.TestReport", report)) + + assert rampart_session.is_incomplete is True + assert rampart_session._results == [] diff --git a/tests/unit/pytest_plugin/test_xdist.py b/tests/unit/pytest_plugin/test_xdist.py index 416eea91..1245e4d3 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) @@ -595,6 +682,80 @@ def test_rejects_legacy_schema_version(self) -> None: 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"): @@ -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], 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.