diff --git a/docs/api/core-protocols.md b/docs/api/core-protocols.md index c4571992..dc9c3d5c 100644 --- a/docs/api/core-protocols.md +++ b/docs/api/core-protocols.md @@ -55,6 +55,16 @@ Protocols and ABCs that define RAMPART's extension points. Implement these to co - register_default_handler_factory - clear_default_handler_factory +## Trace Execution + +::: rampart.core.trace + options: + members: + - EvaluationRecord + - TraceRun + - run_trace_async + - evaluate_final_trace_async + ## Errors ::: rampart.core.errors diff --git a/docs/api/core-types.md b/docs/api/core-types.md index fffdc3c7..df72c8c6 100644 --- a/docs/api/core-types.md +++ b/docs/api/core-types.md @@ -37,7 +37,6 @@ available from `rampart.core`; established result types remain importable from - resolve_attack_verdict - resolve_probe_verdict - resolve_as_attack - - resolve_as_probe ## Configuration diff --git a/docs/concepts/overview.md b/docs/concepts/overview.md index d738971b..6ac8ba25 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(final trace 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 final-trace 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 53f01cfa..6311e8d9 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 should intentionally end the trace early. + --- ## Available Probes diff --git a/docs/concepts/trace-schema.md b/docs/concepts/trace-schema.md index e633a7b3..2a195380 100644 --- a/docs/concepts/trace-schema.md +++ b/docs/concepts/trace-schema.md @@ -61,8 +61,9 @@ stops, including when the turn budget is reached. why that online evaluation ran. A non-null purpose requires an evaluation on the same turn. `Result.trace_end_reason` records why turn production stopped. These provenance fields are optional: missing or null means the producer did not record -them, not that the last online evaluation is the terminal one. The codec never -infers terminal evidence or a stop reason from the result status or turns. +them, not that the last online evaluation is the final-trace evaluation. The +codec never infers final-trace evidence or a stop reason from the result status +or turns. Both placements of `EvalResult` receive the same strict type, finite-confidence, Unicode-scalar, and closed-enum validation. diff --git a/docs/contributing/architecture.md b/docs/contributing/architecture.md index e1f58c02..d3b81292 100644 --- a/docs/contributing/architecture.md +++ b/docs/contributing/architecture.md @@ -42,7 +42,7 @@ When adding a new attack or probe, you add a static factory method — not a new Evaluators are **polarity-free**. They report whether a condition was detected, not whether it's good or bad. The attack/probe factory applies the correct polarity: - `resolve_as_attack`: detected → UNSAFE -- `resolve_as_probe`: detected → SAFE +- `resolve_probe_verdict`: detected → SAFE This allows the same evaluator (e.g., `ToolCalled`) to be used in both attack and probe contexts. diff --git a/docs/contributing/extending-rampart.md b/docs/contributing/extending-rampart.md index 35883f76..eace6255 100644 --- a/docs/contributing/extending-rampart.md +++ b/docs/contributing/extending-rampart.md @@ -181,29 +181,48 @@ 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): +```python +from rampart.core import ( + SafetyStatus, + evaluate_final_trace_async, + resolve_probe_verdict, + run_trace_async, +) -- return "my_attack" -+ return "my_probe" +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_final_trace_async( + evaluator=self._evaluator, + run=run, + ) -- status = resolve_as_attack(eval_results=eval_results) -+ status = resolve_as_probe(eval_results=eval_results) +status = ( + SafetyStatus.ERROR + if evaluation is None + else 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 `final_trace_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 +233,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..013c9188 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 final-trace evaluation (detected → SAFE, not detected → UNSAFE) ### Testing a New Evaluator diff --git a/docs/probes/behavioral.md b/docs/probes/behavioral.md index a3b0e23e..d9488c1d 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. @@ -80,6 +81,27 @@ result = await Probes.behavior( be ignored. Scope applies only to turns in the evaluator context; it does not force an execution to produce every planned turn. + 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. + +!!! note "Upgrading from per-turn probe verdicts" + Earlier releases evaluated a probe after each response, stopped at the + first detection, and combined the per-turn results. Probes now evaluate the + completed trace once. Single-prompt probes with deterministic evaluators + keep the same verdicts. Multi-turn probes can resolve differently because + the evaluator's scope now applies to the full trace, which runs up to + `max_turns` unless `stop_when` is set. A stochastic evaluator, such as an + LLM judge, is sampled once per run instead of once per turn, so trial pass + rates can shift. Replace `resolve_as_probe(eval_results=...)` with + `resolve_probe_verdict(evaluation=...)`. + --- ## Parameters @@ -92,6 +114,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. | +| `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 diff --git a/docs/usage/authoring-tests.md b/docs/usage/authoring-tests.md index 7749d1c3..16a8f21b 100644 --- a/docs/usage/authoring-tests.md +++ b/docs/usage/authoring-tests.md @@ -167,6 +167,10 @@ example `Pattern found on turn(s): 0, 2`. `CURRENT_TURN` uses the same format with only the latest turn number. A failed `ALL_TURNS` match identifies the missing turns with `Pattern missing on turn(s): ...`. + 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 8fe71530..1d7e6939 100644 --- a/docs/usage/results-and-reporting.md +++ b/docs/usage/results-and-reporting.md @@ -60,16 +60,20 @@ stops, not simply the last online evaluation. 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. +Behavioral probes evaluate the completed trace once by default. Their +`Result.final_trace_evaluation` contains the verdict evidence, while +`Result.turn_evaluations` is normally empty. Configure `stop_when` only when online stop evidence is +intentionally needed. + +Strategies that do not record a final-trace verdict leave +`final_trace_evaluation` and `trace_end_reason` as `None`; manually constructed +and error results may do the same intentionally. Online evaluations attached to turns are available as -`result.turn_evaluations`; this list excludes the terminal evaluation. +`result.turn_evaluations`; this list excludes the final-trace evaluation. The former `result.eval_results` property has been removed. Use `result.turn_evaluations` for online evidence and `result.final_trace_evaluation` -for terminal verdict evidence. +for final-trace verdict evidence. `TraceEndReason.MAX_TURNS_REACHED` records budget truncation. It does not by itself claim that the scenario reached semantic completion; each execution @@ -80,19 +84,21 @@ 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: +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 each evaluation's `undetermined_operands` carries it one reason at a time. Inspect final-trace evidence as well as any online evaluations when choosing to fail on a gap: ```python +evaluations = result.turn_evaluations +if result.final_trace_evaluation is not None: + evaluations.append(result.final_trace_evaluation) gaps = [ reason - for turn in result.turns - if turn.eval_result is not None - for reason in turn.eval_result.undetermined_operands + for evaluation in evaluations + for reason in evaluation.undetermined_operands ] assert result and not gaps, result.summary ``` -`JsonFileReportSink` writes the same list as `eval_undetermined_operands` on each turn that has one, and omits the key otherwise. A failing run can carry the key too, so read it alongside `status`: together they tell a fully observed pass from one reached with a gap. No counter makes that distinction, because a qualified pass lands in `safe_count` like any other. +`JsonFileReportSink` writes final-trace gaps as `final_trace_evaluation.undetermined_operands` and online gaps as `eval_undetermined_operands` on each turn. Empty gap lists are omitted. A failing run can carry these keys too, so read them alongside `status`: together they tell a fully observed pass from one reached with a gap. No counter makes that distinction, because a qualified pass lands in `safe_count` like any other. XPIA applies one further rule of its own to `RESPONSE_ONLY` adapters, which does move the verdict. See [Observability Adjustment](../attacks/xpia.md#observability-adjustment). diff --git a/rampart/__init__.py b/rampart/__init__.py index 47d5fde1..f401193b 100644 --- a/rampart/__init__.py +++ b/rampart/__init__.py @@ -35,7 +35,6 @@ Result, SafetyStatus, resolve_as_attack, - resolve_as_probe, ) from rampart.core.types import ( EvalContext, @@ -111,7 +110,6 @@ "execute_trials_async", "record_result", "resolve_as_attack", - "resolve_as_probe", ] diff --git a/rampart/core/__init__.py b/rampart/core/__init__.py index 6487d084..f5752cd4 100644 --- a/rampart/core/__init__.py +++ b/rampart/core/__init__.py @@ -32,10 +32,15 @@ Result, SafetyStatus, resolve_as_attack, - resolve_as_probe, resolve_attack_verdict, resolve_probe_verdict, ) +from rampart.core.trace import ( + EvaluationRecord, + TraceRun, + evaluate_final_trace_async, + run_trace_async, +) from rampart.core.types import ( EvalContext, EvalOutcome, @@ -63,6 +68,7 @@ "EvalOutcome", "EvalResult", "EvaluationPurpose", + "EvaluationRecord", "Evaluator", "ExecutionEvent", "ExecutionEventData", @@ -92,11 +98,13 @@ "ToolCall", "ToolDeclaration", "TraceEndReason", + "TraceRun", "Turn", + "evaluate_final_trace_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/result.py b/rampart/core/result.py index 5981b211..9820c75c 100644 --- a/rampart/core/result.py +++ b/rampart/core/result.py @@ -4,7 +4,7 @@ """Core result types for the RAMPART framework. Defines single-run and population result types, SafetyStatus, HarmCategory, -InjectionRecord, and the resolve_as_attack / resolve_as_probe functions that +InjectionRecord, and the resolve_attack_verdict / resolve_probe_verdict functions that map evaluator outcomes to safety verdicts. Also holds the private helpers that word the undetermined parts of a summary, which execution strategies share. """ @@ -161,12 +161,12 @@ class Result: ``adapter.observability_profile``. final_trace_evaluation: Evaluator output for the final 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. + None for manual/error results and execution strategies that do + not record a final-trace verdict. 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. + strategy does not record 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 @@ -330,7 +330,7 @@ def resolve_attack_verdict(*, evaluation: EvalResult) -> SafetyStatus: """Map one final evaluation using attack polarity. Args: - evaluation: The terminal-trace evaluator result. + evaluation: The final-trace evaluator result. Returns: SafetyStatus: DETECTED maps to UNSAFE, NOT_DETECTED maps to SAFE, @@ -351,7 +351,7 @@ def resolve_probe_verdict(*, evaluation: EvalResult) -> SafetyStatus: """Map one final evaluation using probe polarity. Args: - evaluation: The terminal-trace evaluator result. + evaluation: The final-trace evaluator result. Returns: SafetyStatus: DETECTED maps to SAFE, NOT_DETECTED maps to UNSAFE, @@ -414,33 +414,6 @@ def resolve_as_attack(*, eval_results: list[EvalResult]) -> SafetyStatus: return SafetyStatus.SAFE -def resolve_as_probe(*, eval_results: list[EvalResult]) -> SafetyStatus: - """Probe semantics: detected -> SAFE, not detected -> UNSAFE. - - Shared by all probe execution strategies. - - Precedence: NOT_DETECTED > UNDETERMINED > DETECTED. If any evaluator - failed to detect the expected behavior, the agent is provably - non-compliant regardless of whether other evaluators were undetermined. - UNDETERMINED only matters when no evaluator produced a definitive - negative signal. - - Args: - eval_results: List of evaluator outcomes. - - Returns: - SafetyStatus: The resolved status. - """ - 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): - return SafetyStatus.UNDETERMINED - return SafetyStatus.SAFE - - def _summarize_undetermined_operands(*, eval_results: list[EvalResult]) -> str: """Describe the parts of an evaluation that never reached a determination. diff --git a/rampart/core/trace.py b/rampart/core/trace.py new file mode 100644 index 00000000..0741889c --- /dev/null +++ b/rampart/core/trace.py @@ -0,0 +1,229 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Shared linear trace execution and final-trace evaluation helpers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from typing import TYPE_CHECKING + +from rampart.common.text import safe_str_list +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, + ) + + +def _matches_final_trace_context(*, context: EvalContext, run: TraceRun) -> bool: + """Check that the raw trace and adapter context are unchanged. + + Returns: + bool: Whether this context can supply the final-trace judgment. + """ + return ( + context.observability_level is run.observability_level + and context.manifest is run.manifest + and len(context.turns) == len(run.raw_turns) + and all( + evaluated is final + for evaluated, final in zip(context.turns, run.raw_turns, strict=True) + ) + ) + + +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_final_trace_async( + *, + evaluator: Evaluator, + run: TraceRun, +) -> EvalResult | None: + """Evaluate the final 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. Reuse requires matching evaluator, raw-turn, and manifest + identities and the same observability level. Requests, responses, manifests, + and their nested values are treated as immutable once evaluated. + """ + if not run.raw_turns: + return None + + record = run.latest_online_evaluation + if ( + record is not None + and record.evaluator is evaluator + and _matches_final_trace_context(context=record.context, run=run) + ): + return replace( + record.result, + evidence=safe_str_list(value=record.result.evidence), + undetermined_operands=safe_str_list( + value=record.result.undetermined_operands, + ), + ) + + 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/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 271b14ec..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,6 +73,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. + 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 @@ -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 eb010b11..a8afc514 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, final-trace 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_final_trace_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,6 +49,8 @@ class SingleTurnExecution(BaseExecution): Args: driver (PromptDriver): How to drive the conversation. evaluator (Evaluator): What behavior to check for. + 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 @@ -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_final_trace_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, + ), + final_trace_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 final-trace 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 4a5e5884..40f1f81a 100644 --- a/rampart/pytest_plugin/_xdist.py +++ b/rampart/pytest_plugin/_xdist.py @@ -55,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 @@ -494,7 +494,7 @@ def _serialize_population_ref( def _serialize_result(*, result: Result, nodeid: str) -> dict[str, Any]: """Serialize a Result to a JSON-safe dict for the xdist transport. - This full-fidelity transport projection round-trips terminal and online + This full-fidelity transport projection round-trips final-trace 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 diff --git a/schemas/trace-compatibility.json b/schemas/trace-compatibility.json index 0d0ddf13..aa953b23 100644 --- a/schemas/trace-compatibility.json +++ b/schemas/trace-compatibility.json @@ -1,8 +1,8 @@ { "version": "rampart.trace.v2", - "contract_sha256": "828c2783d81d06185caeeef1a7ff83e025d9470d3c9dc894a6a555e555c47ed9", - "previous_contract_sha256": "da1a36daa27b5fc2cdaf4305ba3aa743f52789643de94ed97916ebbaa5fc2b47", - "decision": "new-major", - "rationale": "V2 requires nonempty PopulationRef.id values, rejecting records accepted by v1 and therefore requiring a major version bump. The optional final_trace_evaluation, trace_end_reason, and Turn.eval_purpose fields define absence as not recorded; those additions alone are compatible. Repeated EvalResult definitions now consistently receive canonical validation policies. The published v1 schema is preserved unchanged.", - "migration_note": "This reader accepts only v2; no v1 decoder, upcaster, or migration utility is provided. Convert v1 records explicitly into separate v2 records and validate them with deserialize_record(); do not merely relabel their version. Empty population IDs require legitimate producer provenance or regeneration, not invented identifiers. Leave unrecorded final_trace_evaluation, trace_end_reason, and eval_purpose fields absent or null rather than inferring them from online evaluations. Preserve the original records." + "contract_sha256": "f3cc87fce1006b240a376b167ad1d874b6219c9aaea23a30063a0b4c232edb1a", + "previous_contract_sha256": "828c2783d81d06185caeeef1a7ff83e025d9470d3c9dc894a6a555e555c47ed9", + "decision": "compatible", + "rationale": "Removes the list-based resolve_as_probe helper and updates Result documentation. Result fields, canonical serialization, and published schemas are unchanged, so existing v2 records remain valid.", + "migration_note": null } diff --git a/tests/unit/core/test_result.py b/tests/unit/core/test_result.py index c5d16990..95989225 100644 --- a/tests/unit/core/test_result.py +++ b/tests/unit/core/test_result.py @@ -18,7 +18,6 @@ _explain_undetermined, _summarize_undetermined_operands, resolve_as_attack, - resolve_as_probe, resolve_attack_verdict, resolve_probe_verdict, ) @@ -352,7 +351,7 @@ def test_accepts_large_but_semantically_valid_provenance(self) -> None: class TestResultTurnEvaluationsProperty: - """Turn evaluations remain separate from the terminal evaluation.""" + """Turn evaluations remain separate from the final-trace evaluation.""" def test_removed_eval_results_property_is_absent(self) -> None: assert not hasattr(_result(SafetyStatus.SAFE), "eval_results") @@ -487,64 +486,6 @@ def test_rejects_malformed_runtime_outcome(self) -> None: resolve_as_attack(eval_results=[malformed]) -class TestResolveAsProbe: - def test_empty_returns_error(self) -> None: - status = resolve_as_probe(eval_results=[]) - assert status is SafetyStatus.ERROR - - def test_detected_returns_safe(self) -> None: - status = resolve_as_probe( - eval_results=[_er(EvalOutcome.DETECTED)], - ) - assert status is SafetyStatus.SAFE - - def test_not_detected_returns_unsafe(self) -> None: - status = resolve_as_probe( - eval_results=[_er(EvalOutcome.NOT_DETECTED)], - ) - assert status is SafetyStatus.UNSAFE - - def test_undetermined_only_returns_undetermined(self) -> None: - status = resolve_as_probe( - eval_results=[_er(EvalOutcome.UNDETERMINED)], - ) - assert status is SafetyStatus.UNDETERMINED - - def test_mixed_not_detected_and_undetermined_returns_unsafe(self) -> None: - status = resolve_as_probe( - eval_results=[ - _er(EvalOutcome.UNDETERMINED), - _er(EvalOutcome.NOT_DETECTED), - ], - ) - assert status is SafetyStatus.UNSAFE - - def test_mixed_detected_and_undetermined_returns_undetermined(self) -> None: - status = resolve_as_probe( - eval_results=[ - _er(EvalOutcome.DETECTED), - _er(EvalOutcome.UNDETERMINED), - ], - ) - assert status is SafetyStatus.UNDETERMINED - - def test_all_detected_returns_safe(self) -> None: - status = resolve_as_probe( - eval_results=[ - _er(EvalOutcome.DETECTED), - _er(EvalOutcome.DETECTED), - ], - ) - 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: clause = _summarize_undetermined_operands( @@ -842,6 +783,14 @@ def test_rejects_malformed_runtime_outcome(self) -> None: class TestResolveProbeVerdict: + def test_aggregate_resolver_is_not_exported(self) -> None: + import rampart + from rampart import core + from rampart.core import result + + for module in (rampart, core, result): + assert not hasattr(module, "resolve_as_probe") + @pytest.mark.parametrize( ("evaluation", "expected"), [ diff --git a/tests/unit/core/test_trace.py b/tests/unit/core/test_trace.py new file mode 100644 index 00000000..4e0b858b --- /dev/null +++ b/tests/unit/core/test_trace.py @@ -0,0 +1,398 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for the shared linear trace runner.""" + +from __future__ import annotations + +from dataclasses import replace +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_final_trace_async, run_trace_async +from rampart.core.types import ( + EvalOutcome, + EvalResult, + EvaluationPurpose, + ObservabilityLevel, + Request, + Response, + TraceEndReason, + Turn, +) +from rampart.drivers.static import StaticDriver +from rampart.evaluators import ToolCalled +from tests.fixtures import MockSession + + +class _FailingIterable: + def __iter__(self) -> object: + raise RuntimeError("evidence unavailable") + + +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 TestEvaluateFinalTraceAsync: + async def test_changed_observability_requires_new_judgment_async(self) -> None: + evaluator = ToolCalled("send_email") + 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, + ) + updated_run = replace( + run, + observability_level=ObservabilityLevel.RESPONSE_ONLY, + ) + + result = await evaluate_final_trace_async(evaluator=evaluator, run=updated_run) + + assert run.latest_online_evaluation is not None + assert run.latest_online_evaluation.result.outcome is EvalOutcome.NOT_DETECTED + assert result is not None + assert result.outcome is EvalOutcome.UNDETERMINED + + @pytest.mark.parametrize("manifest", [None, AppManifest(name="agent")]) + async def test_changed_manifest_requires_new_judgment_async( + self, + manifest: AppManifest | None, + ) -> None: + evaluator = _evaluator(EvalOutcome.NOT_DETECTED, EvalOutcome.DETECTED) + 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, + manifest=AppManifest(name="agent"), + ) + + result = await evaluate_final_trace_async( + evaluator=evaluator, + run=replace(run, manifest=manifest), + ) + + assert result is not None and result.outcome is EvalOutcome.DETECTED + assert evaluator.evaluate_async.await_count == 2 + assert ( + evaluator.evaluate_async.await_args.kwargs["context"].manifest is manifest + ) + + @pytest.mark.parametrize( + ("evidence", "expected"), + [ + (None, []), + (42, []), + (_FailingIterable(), []), + ("confirmed", ["confirmed"]), + (["confirmed"], ["confirmed"]), + ], + ) + async def test_reuse_preserves_verdict_with_optional_evidence_async( + self, + evidence: object, + expected: list[str], + ) -> None: + online = EvalResult( + outcome=EvalOutcome.DETECTED, + evidence=evidence, # ty: ignore[invalid-argument-type] + rationale="condition confirmed", + undetermined_operands=["missing side effects"], + ) + evaluator = AsyncMock(spec=Evaluator) + evaluator.evaluate_async.return_value = online + 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, + ) + + result = await evaluate_final_trace_async(evaluator=evaluator, run=run) + + assert result is not None + assert result.outcome is EvalOutcome.DETECTED + assert result.rationale == online.rationale + assert result.evidence == expected + assert result.evidence is not online.evidence + assert result.undetermined_operands == online.undetermined_operands + assert result.undetermined_operands is not online.undetermined_operands + evaluator.evaluate_async.assert_awaited_once() + + 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_final_trace_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, + manifest=AppManifest(name="agent"), + ) + online_result = run.latest_online_evaluation + assert online_result is not None + + result = await evaluate_final_trace_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_final_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_final_trace_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_final_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_final_trace_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_final_trace_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/probes/test_single_turn.py b/tests/unit/probes/test_single_turn.py index 0ef19396..9ac927a6 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,6 +24,7 @@ Request, Response, ToolCall, + TraceEndReason, Turn, ) from rampart.drivers.static import StaticDriver @@ -34,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.""" @@ -112,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.""" @@ -367,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.final_trace_evaluation is None + assert result.trace_end_reason is None class TestProbeEndToEnd: @@ -424,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 final-trace evaluation.""" async def test_max_turns_resolves_normally_async(self) -> None: adapter = _adapter(responses=[Response(text="ok")]) @@ -438,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.""" +class TestProbeSummary: + """Final-trace evaluation summaries preserve evidence and observability gaps.""" - def test_summary_uses_only_not_detected_rationales(self) -> None: + def test_unsafe_summary_uses_final_trace_rationale(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 - - def test_summary_falls_back_without_a_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 == ( @@ -612,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" @@ -662,12 +664,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" @@ -675,13 +676,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'" @@ -693,12 +693,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: " @@ -706,12 +705,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: " @@ -730,12 +728,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" @@ -743,12 +740,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: " @@ -763,12 +759,322 @@ 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.final_trace_evaluation is evaluator.evaluate_async.return_value + assert result.turn_evaluations == [] + 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.final_trace_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="final 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.final_trace_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_current_turn_scope_checks_only_final_probe_response_async( + self, + ) -> None: + result = await Probes.behavior( + prompts=["p1", "p2"], + evaluator=ResponseContains("ok", scope=ResponseScope.CURRENT_TURN), + ).execute_async( + adapter=_adapter( + responses=[Response(text="not yet"), Response(text="ok")], + ), + ) + + assert len(result.turns) == 2 + assert result.status is SafetyStatus.SAFE + assert result.final_trace_evaluation is not None + assert result.final_trace_evaluation.evidence == ["Pattern found on turn(s): 1"] + + async def test_final_trace_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_final_trace_evidence_async(self) -> None: + evaluator = AsyncMock() + evaluator.evaluate_async.return_value = EvalResult( + outcome=EvalOutcome.DETECTED, + evidence=["final-trace evidence"], + rationale="final-trace rationale", + ) + + result = await Probes.behavior( + prompt="p", + evaluator=evaluator, + ).execute_async(adapter=_adapter(responses=[Response(text="r")])) + + assert "final-trace evidence" in result.summary + + async def test_undetermined_summary_includes_final_trace_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 aa2dfce4..7b89a641 100644 --- a/tests/unit/pytest_plugin/test_xdist.py +++ b/tests/unit/pytest_plugin/test_xdist.py @@ -673,9 +673,9 @@ 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": [], } @@ -1140,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, }, } @@ -1541,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: