diff --git a/docs/api/core-protocols.md b/docs/api/core-protocols.md index c457199..dc9c3d5 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/rampart/core/__init__.py b/rampart/core/__init__.py index 6487d08..89a5032 100644 --- a/rampart/core/__init__.py +++ b/rampart/core/__init__.py @@ -36,6 +36,12 @@ 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 +69,7 @@ "EvalOutcome", "EvalResult", "EvaluationPurpose", + "EvaluationRecord", "Evaluator", "ExecutionEvent", "ExecutionEventData", @@ -92,11 +99,14 @@ "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/trace.py b/rampart/core/trace.py new file mode 100644 index 0000000..0741889 --- /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/tests/unit/core/test_trace.py b/tests/unit/core/test_trace.py new file mode 100644 index 0000000..4e0b858 --- /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, + )