diff --git a/README.md b/README.md index 05663e8..44f6ddb 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,8 @@

Quick start · Assertions · + Guardrails · + Evals · Reports · Full guide · E2E flow @@ -44,7 +46,7 @@ Tests run against mocks by default. Live tests are opt-in. See Quick start to te ## The report is the product -Most agent tests dump JSON into CI logs nobody reads. **agent-test-kit** generates a single self-contained HTML file — a GO / NO-GO readiness banner, tool-flow strips, MCP chips, prompt review diffs, golden guard outcomes — the same UI whether you ran mock tests or live E2E. +Most agent tests dump JSON into CI logs nobody reads. **agent-test-kit** generates a single self-contained HTML file — a GO / NO-GO readiness banner, tool-flow strips, MCP chips, prompt review diffs, golden guard outcomes, and guardrail and eval results — the same UI whether you ran mock tests or live E2E. ``` ┌─ Agent test report ─────────────────────────────────────────────┐ @@ -62,7 +64,7 @@ Most agent tests dump JSON into CI logs nobody reads. **agent-test-kit** generat │ │ get_inv… │ │ get_iss… │ │ create… │ │ │ └──────────┘ └──────────┘ └──────────┘ │ ├─────────────────────────────────────────────────────────────────┤ -│ ▾ Assertions ▾ Tool calls ▾ Timeline ▾ Side effects │ +│ ▾ Assertions ▾ Guardrails ▾ Evals ▾ Tool calls ▾ Timeline │ └─────────────────────────────────────────────────────────────────┘ ``` @@ -84,7 +86,7 @@ Redaction is on by default — tokens, credentials, and sensitive args never hit **Your agent repo owns:** test cases, prompts, fixtures, domain verifiers, cleanup hooks, and any LLM judge. -**This package provides:** HTTP/Cursor clients, normalized traces, workflow assertions, deterministic behavior scorers, goal-based regression cases, JSON/HTML reports with a GO / NO-GO readiness block, and verifier orchestration. +**This package provides:** HTTP/Cursor clients, normalized traces, workflow assertions, offline and online guardrails, evals, deterministic behavior scorers, goal-based regression cases, JSON/HTML reports with a GO / NO-GO readiness block, and verifier orchestration. The pytest plugin loads automatically when installed — no `pytest_plugins` line needed. @@ -130,6 +132,7 @@ markers = agent_integration: deployed agent, read-only agent_e2e: live HTTP (opt-in) agent_write_action: real writes (opt-in) + agent_eval: registered evals selected by name addopts = -m "not agent_e2e and not agent_write_action" ``` @@ -369,7 +372,159 @@ class GroundednessJudge: await result.assert_score(GroundednessJudge(), min_score=0.8) ``` -### Regression cases (YAML-driven) +## Guardrails + +A guardrail checks assistant output and tool flow. The same object runs offline, against a recorded trace, and online, against a live stream. Message and tool-call hooks run as each event arrives. Trace hooks, such as a required final tool, run when the run finishes. + +Subclass `Guardrail` or use `@guardrail`. Return a violation message, or `None` when the event is allowed. + +| Guardrail | What it checks | +|-----------|----------------| +| `NoProfanityGuardrail(terms=None, *, extra_terms=())` | Disallowed language in assistant output. Defaults to a small profanity list. `terms` replaces that list; `extra_terms` adds words or phrases. | +| `RequiredFinalToolGuardrail(tool_name, *, server=None)` | The last tool call is `tool_name` and succeeded. | +| `must_open_pull_request(tool_name="create_pull_request", *, server=None)` | Flow rule: the run ends by opening a pull request. | + +Failures are `AssertionError`s. They are stored on the result and rendered in the HTML report under **Guardrails**, including when the test fails. + +### Offline guardrail + +```python +import pytest +from agent_test_kit import ( + AgentExecutionResult, + NoProfanityGuardrail, + ToolCall, + must_open_pull_request, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace + + +@pytest.mark.agent_unit +def test_recorded_run_stays_within_guardrails(agent_guardrails): + result = AgentExecutionResult( + success=True, + run_id="run_1", + response="I opened the pull request.", + trace=AgentTrace( + tool_calls=[ + ToolCall( + server="github", + name="create_pull_request", + status=ToolCallStatus.SUCCESS, + operation_kind=ToolOperationKind.WRITE, + ) + ] + ), + ) + agent_guardrails.check( + result, + [NoProfanityGuardrail(), must_open_pull_request(server="github")], + ) +``` + +`check_guardrails(result, guardrails)` is the same check without pytest. `result.assert_guardrails(...)` records it like the other assertions. + +### Online guardrail + +`GuardrailMonitor` is the streaming hook. Call `handle` (or `on_message` / `on_tool_call`) for each event. `fail_fast=True` raises `GuardrailViolationError` at the first violation so middleware can stop the run. Leave it false to collect every violation, then call `report.assert_passed()`. + +```python +from agent_test_kit import GuardrailMonitor, NoProfanityGuardrail, must_open_pull_request + + +def test_live_stream_opens_pr(agent_guardrails): + monitor = agent_guardrails.monitor( + [NoProfanityGuardrail(), must_open_pull_request(server="github")], + fail_fast=True, + ) + + def on_agent_event(event: dict) -> None: + monitor.handle(event) + + on_agent_event({"type": "message", "text": "I opened the pull request."}) + on_agent_event({ + "type": "tool_call", + "server": "github", + "name": "create_pull_request", + "status": "success", + "operation_kind": "write", + }) + monitor.finish().assert_passed() +``` + +Outside pytest, construct `GuardrailMonitor` yourself. The fixture records the online result on teardown, including when a violation aborts the test. + +### Write your own + +```python +from agent_test_kit import Guardrail, ToolCall, guardrail + + +@guardrail(phase="message") +def no_secrets(text: str, *, role: str = "assistant") -> str | None: + if role == "assistant" and "sk-" in text: + return "assistant output looks like a secret key" + return None + + +class NoDeleteGuardrail(Guardrail): + name = "no_delete" + + def on_tool_call(self, call: ToolCall) -> str | None: + if call.name.startswith("delete_"): + return f"delete tool is not allowed: {call.name}" + return None +``` + +`phase` is `"message"`, `"tool_call"`, or `"trace"`. A decorated guardrail is a `Guardrail`, so it works with `check_guardrails` and `GuardrailMonitor`. + +## Evals + +An eval scores one run from 0.0 to 1.0 and passes when the score is at least its threshold. Register it with `@eval_case`. Run it with the `agent_eval` fixture, or call `run_evals` / `result.assert_evals` directly. `@pytest.mark.agent_eval("name")` selects registered evals by name when you call `agent_eval.run(result)` without a list. + +| Eval | Score | +|------|-------| +| `RequiredTermsEval(terms, *, threshold=1.0)` | Fraction of terms present in the response | +| `RequiredToolsEval(tools, *, threshold=1.0)` | Fraction of required tools that were called. A tool is a name or `(server, name)`. | + +```python +import pytest +from agent_test_kit import AgentExecutionResult, RequiredTermsEval, eval_case, response_text + + +@eval_case(threshold=1.0) +def mentions_pull_request(result) -> float: + return 1.0 if "pull request" in response_text(result).casefold() else 0.0 + + +@pytest.mark.agent_eval("mentions_pull_request") +@pytest.mark.agent_unit +def test_mentions_pull_request(agent_eval): + result = AgentExecutionResult( + success=True, + run_id="run_1", + response="I opened the pull request.", + ) + agent_eval.run(result) + + +def test_terms_with_an_explicit_eval(agent_eval): + result = AgentExecutionResult(success=True, run_id="run_2", response="refund for order 123") + agent_eval.run(result, [RequiredTermsEval(["refund", "123"])]) +``` + +`run_evals(result, [mentions_pull_request])` is the same call outside pytest. Scores, thresholds, and pass/fail appear in the HTML report under **Evals**. A failing eval raises `AssertionError` and is still recorded. + +Subclass `Eval` and implement `score` to return a `Score`, a float, or a bool. `register_eval` adds it under `evaluation.name`. + +Runnable copies of the offline guardrail, the online guardrail, and an eval: [`examples/guardrails_and_evals.py`](examples/guardrails_and_evals.py). + +```bash +python examples/guardrails_and_evals.py +``` + +## Regression cases (YAML-driven) Load cases with `load_regression_cases(path)`; evaluate with `case.evaluate(result)`. @@ -435,7 +590,7 @@ for case in load_regression_cases("tests/data/behavior_cases.yaml"): `case.run()` executes `runs` times with one idempotency key (`regression:` by default) and checks that every scorer named in `behavior.scores` was provided before calling the agent. To score results you already have, use `await case.evaluate_runs(results, scorers=...)`. Full example: [`examples/behavior_regression_cases.yaml`](examples/behavior_regression_cases.yaml). -### Typical PR-review workflow (all together) +## Typical PR-review workflow ```python result.assert_success() @@ -559,7 +714,9 @@ See [First test](#first-test). More examples: [`examples/consumer_test_example.p | `agent_test_config` | `AgentTestConfig` from env / defaults | | `agent_client` | `AgentClient` + auto report attachment | | `cursor_agent_client` | `CursorAgentClient` + auto report attachment | -| `agent_scenario` | Attach results, verifiers, and `attach_regression_evaluation` for persona readiness | +| `agent_scenario` | Attach results, verifiers, guardrail and eval reports, and `attach_regression_evaluation` | +| `agent_guardrails` | Offline `check` and online `monitor` for guardrails | +| `agent_eval` | Run evals passed in, or named by `@pytest.mark.agent_eval` | | `agent_cleanup` | Async cleanup manager (runs on teardown) | ```python @@ -721,6 +878,7 @@ Reference examples in **this repo**: [`examples/consumer_test_example.py`](examp | `agent_regression` | Regression case files | | `agent_security` | Safety / redaction checks | | `agent_multi_mcp` | Multi-MCP orchestration | +| `agent_eval` | Selects registered evals for the `agent_eval` fixture | --- @@ -746,8 +904,12 @@ from agent_test_kit import ( AgentExecutionResult, AnyOfStep, CleanupManager, + Eval, ForbiddenTermsScorer, + Guardrail, + GuardrailMonitor, HtmlReportWriter, + NoProfanityGuardrail, JsonReportWriter, OptionalStep, PromptReviewClient, @@ -755,15 +917,22 @@ from agent_test_kit import ( RegressionRunEvaluation, RepeatedExecutionResult, RequiredFieldsScorer, + RequiredTermsEval, RequiredTermsScorer, + RequiredToolsEval, Score, Scorer, + ToolCall, ToolStep, + check_guardrails, + eval_case, evaluate_readiness, get_profile, load_regression_cases, load_report, + must_open_pull_request, response_text, + run_evals, run_repeatedly, run_verifiers, SideEffectVerifier, diff --git a/examples/guardrails_and_evals.py b/examples/guardrails_and_evals.py new file mode 100644 index 0000000..0e7f7fe --- /dev/null +++ b/examples/guardrails_and_evals.py @@ -0,0 +1,104 @@ +"""Runnable guardrail and eval examples. + +From a checkout with the package installed: + + python examples/guardrails_and_evals.py +""" + +from __future__ import annotations + +from agent_test_kit import ( + AgentExecutionResult, + EvalResult, + GuardrailMonitor, + GuardrailReport, + NoProfanityGuardrail, + RequiredTermsEval, + ToolCall, + check_guardrails, + eval_case, + must_open_pull_request, + response_text, + run_evals, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace + + +def _pr_result(response: str) -> AgentExecutionResult: + return AgentExecutionResult( + success=True, + run_id="run_example", + response=response, + trace=AgentTrace( + tool_calls=[ + ToolCall( + server="github", + name="create_pull_request", + status=ToolCallStatus.SUCCESS, + operation_kind=ToolOperationKind.WRITE, + ) + ] + ), + ) + + +def example_offline_guardrail() -> GuardrailReport: + """Check a recorded trace the same way a pytest test would.""" + result = _pr_result("I opened the pull request.") + report = check_guardrails( + result, + [NoProfanityGuardrail(), must_open_pull_request(server="github")], + ) + report.assert_passed() + return report + + +def example_online_guardrail() -> GuardrailReport: + """Attach the same guardrails to a live stream of messages and tool calls.""" + monitor = GuardrailMonitor( + [NoProfanityGuardrail(), must_open_pull_request(server="github")], + fail_fast=True, + ) + + def on_agent_event(event: dict[str, object]) -> None: + monitor.handle(event) + + on_agent_event({"type": "message", "role": "assistant", "text": "I opened the pull request."}) + on_agent_event( + { + "type": "tool_call", + "server": "github", + "name": "create_pull_request", + "status": "success", + "operation_kind": "write", + } + ) + report = monitor.finish() + report.assert_passed() + return report + + +@eval_case(name="example_mentions_pull_request", threshold=1.0) +def example_mentions_pull_request(result: AgentExecutionResult) -> float: + """1.0 when the assistant mentions the pull request it opened.""" + text = response_text(result).casefold() + return 1.0 if "pull request" in text else 0.0 + + +def example_eval() -> EvalResult: + """Register an eval and score a recorded run.""" + result = _pr_result("I opened the pull request.") + report = run_evals( + result, + [example_mentions_pull_request, RequiredTermsEval(["pull request"])], + ) + report.assert_passed() + return report.results[0] + + +if __name__ == "__main__": + example_offline_guardrail() + example_online_guardrail() + example_eval() + print("guardrail and eval examples passed") diff --git a/pyproject.toml b/pyproject.toml index 87d2161..69d6fb4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -93,6 +93,7 @@ markers = [ "agent_write_action: Tests that perform write actions in target systems", "agent_nondeterministic: Tests that may flake due to model variance", "agent_multi_mcp: Multi-MCP server workflow tests", + "agent_eval: Registered agent evals selected by name for the agent_eval fixture", ] [tool.coverage.run] diff --git a/src/agent_test_kit/__init__.py b/src/agent_test_kit/__init__.py index 1e475cf..ca85af6 100644 --- a/src/agent_test_kit/__init__.py +++ b/src/agent_test_kit/__init__.py @@ -13,8 +13,35 @@ get_profile, ) from agent_test_kit.client.prompt_review_client import PromptReviewClient +from agent_test_kit.evals import ( + Eval, + EvalReport, + EvalResult, + RequiredTermsEval, + RequiredToolsEval, + eval_case, + get_eval, + register_eval, + registered_evals, + run_evals, +) +from agent_test_kit.guardrails import ( + Guardrail, + GuardrailCheck, + GuardrailEvent, + GuardrailMonitor, + GuardrailReport, + GuardrailViolation, + GuardrailViolationError, + NoProfanityGuardrail, + RequiredFinalToolGuardrail, + check_guardrails, + guardrail, + must_open_pull_request, +) from agent_test_kit.models.execution import AgentExecutionResult, AgentFlowResult, AssertionOutcome from agent_test_kit.models.run_context import RunContext +from agent_test_kit.models.tool_call import ToolCall from agent_test_kit.publishing import ( HttpResultPublisher, PublishResult, @@ -67,13 +94,24 @@ "AssertionOutcome", "CleanupManager", "CursorAgentClient", + "Eval", + "EvalReport", + "EvalResult", "FlowAssertionEngine", "ForbiddenTermsScorer", + "Guardrail", + "GuardrailCheck", + "GuardrailEvent", + "GuardrailMonitor", + "GuardrailReport", + "GuardrailViolation", + "GuardrailViolationError", "HtmlReportWriter", "HttpResultPublisher", "JsonReportWriter", "MarkdownStatusReportWriter", "OptionalStep", + "NoProfanityGuardrail", "PromptReviewClient", "PromptAiHelperProfile", "PublishResult", @@ -87,7 +125,10 @@ "RegressionToolExpectation", "RepeatedExecutionResult", "RequiredFieldsScorer", + "RequiredFinalToolGuardrail", + "RequiredTermsEval", "RequiredTermsScorer", + "RequiredToolsEval", "ResultPublishError", "ResultPublisher", "ResultPublisherConfig", @@ -95,16 +136,25 @@ "Score", "Scorer", "SideEffectVerifier", + "ToolCall", "ToolStep", "VerificationContext", "VerificationResult", "VerificationSummary", "__version__", + "check_guardrails", + "eval_case", "evaluate_readiness", + "get_eval", + "guardrail", "load_regression_cases", "load_report", "get_profile", + "must_open_pull_request", + "register_eval", + "registered_evals", "response_text", + "run_evals", "run_repeatedly", "run_verifiers", ] diff --git a/src/agent_test_kit/evals/__init__.py b/src/agent_test_kit/evals/__init__.py new file mode 100644 index 0000000..3a98d28 --- /dev/null +++ b/src/agent_test_kit/evals/__init__.py @@ -0,0 +1,26 @@ +"""Score-based evals with registration and pytest integration.""" + +from agent_test_kit.evals.base import ( + Eval, + EvalReport, + EvalResult, + eval_case, + get_eval, + register_eval, + registered_evals, + run_evals, +) +from agent_test_kit.evals.builtin import RequiredTermsEval, RequiredToolsEval + +__all__ = [ + "Eval", + "EvalReport", + "EvalResult", + "RequiredTermsEval", + "RequiredToolsEval", + "eval_case", + "get_eval", + "register_eval", + "registered_evals", + "run_evals", +] diff --git a/src/agent_test_kit/evals/base.py b/src/agent_test_kit/evals/base.py new file mode 100644 index 0000000..1eefc06 --- /dev/null +++ b/src/agent_test_kit/evals/base.py @@ -0,0 +1,197 @@ +"""Evals: named scores with a pass threshold, registered and run from pytest.""" + +from __future__ import annotations + +import inspect +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field +from typing import Any + +from agent_test_kit.models.execution import AgentExecutionResult +from agent_test_kit.scoring import Score, validate_unit_interval + +_REGISTRY: dict[str, Eval] = {} + + +@dataclass(frozen=True, slots=True) +class EvalResult: + """Score and pass/fail outcome for one eval on one run.""" + + name: str + passed: bool + score: float + threshold: float + reason: str | None = None + details: dict[str, Any] = field(default_factory=dict) + + def assert_passed(self) -> None: + if self.passed: + return + detail = f" ({self.reason})" if self.reason else "" + raise AssertionError( + f"Eval {self.name!r} failed: score {self.score:.2f} " + f"is below {self.threshold:.2f}{detail}" + ) + + +@dataclass(frozen=True, slots=True) +class EvalReport: + """Aggregate of every eval run together.""" + + results: tuple[EvalResult, ...] + + @property + def passed(self) -> bool: + return all(result.passed for result in self.results) + + def assert_passed(self) -> None: + failures = [result for result in self.results if not result.passed] + if not failures: + return + lines = [] + for result in failures: + detail = f" ({result.reason})" if result.reason else "" + lines.append( + f"{result.name}: score {result.score:.2f} is below {result.threshold:.2f}{detail}" + ) + raise AssertionError("Eval failures:\n" + "\n".join(lines)) + + +class Eval: + """Score one execution result. Pass when ``score >= threshold``. + + Subclass and implement ``score``, or build one with ``@eval_case``. + ``score`` may return a ``Score``, a float in ``[0.0, 1.0]``, or a bool. + """ + + name: str = "eval" + threshold: float = 1.0 + + def __init__(self, *, name: str | None = None, threshold: float = 1.0) -> None: + if name is not None: + self.name = name + if not isinstance(self.name, str) or not self.name.strip(): + raise ValueError("eval name must be a non-empty string") + validate_unit_interval(f"eval {self.name} threshold", threshold) + self.threshold = threshold + + def score(self, result: AgentExecutionResult) -> Score | float | bool: + """Return a score for ``result``.""" + raise NotImplementedError + + def evaluate(self, result: AgentExecutionResult) -> EvalResult: + """Score ``result`` and compare it to the threshold.""" + raw = self.score(result) + if inspect.isawaitable(raw): + close = getattr(raw, "close", None) + if close is not None: + close() + raise TypeError( + f"Eval {self.name!r} returned an awaitable. " + "score() must be synchronous and return a Score, float, or bool." + ) + value, reason, details = _coerce_score(self.name, raw) + passed = value >= self.threshold + if reason is None and not passed: + reason = f"score {value:.2f} is below {self.threshold:.2f}" + return EvalResult( + name=self.name, + passed=passed, + score=value, + threshold=self.threshold, + reason=reason, + details=details, + ) + + def __call__(self, result: AgentExecutionResult) -> EvalResult: + return self.evaluate(result) + + +def run_evals( + result: AgentExecutionResult, + evals: Eval | Sequence[Eval], +) -> EvalReport: + """Score ``result`` with each eval. Does not raise; call ``assert_passed``.""" + selected = _as_evals(evals) + return EvalReport(tuple(item.evaluate(result) for item in selected)) + + +def register_eval(evaluation: Eval) -> Eval: + """Register ``evaluation`` under its name. Registering the same object twice is a no-op.""" + if not isinstance(evaluation, Eval): + raise TypeError(f"Expected Eval. Observed: {type(evaluation).__name__}") + existing = _REGISTRY.get(evaluation.name) + if existing is not None and existing is not evaluation: + raise ValueError(f"eval {evaluation.name!r} is already registered") + _REGISTRY[evaluation.name] = evaluation + return evaluation + + +def get_eval(name: str) -> Eval: + """Return a registered eval by name.""" + try: + return _REGISTRY[name] + except KeyError: + known = ", ".join(registered_evals()) or "(none)" + raise KeyError(f"Unknown eval {name!r}. Registered: {known}") from None + + +def registered_evals() -> tuple[str, ...]: + """Names of evals registered with ``@eval_case`` or ``register_eval``.""" + return tuple(sorted(_REGISTRY)) + + +def eval_case( + func: Callable[[AgentExecutionResult], Score | float | bool] | None = None, + *, + name: str | None = None, + threshold: float = 1.0, + register: bool = True, +) -> Any: + """Register a function as an eval. + + The function receives an ``AgentExecutionResult`` and returns a score in + ``[0.0, 1.0]``, a bool, or a ``Score``. The eval passes when the score is + at least ``threshold``. + """ + + def decorate(fn: Callable[[AgentExecutionResult], Score | float | bool]) -> Eval: + label = name or fn.__name__ + + class _FunctionEval(Eval): + def score(self, result: AgentExecutionResult) -> Score | float | bool: + return fn(result) + + _FunctionEval.__name__ = label + _FunctionEval.__doc__ = fn.__doc__ + evaluation = _FunctionEval(name=label, threshold=threshold) + if register: + register_eval(evaluation) + return evaluation + + if func is not None: + return decorate(func) + return decorate + + +def _as_evals(evals: Eval | Sequence[Eval]) -> tuple[Eval, ...]: + selected = (evals,) if isinstance(evals, Eval) else tuple(evals) + if not selected: + raise ValueError("at least one eval is required") + for item in selected: + if not isinstance(item, Eval): + raise TypeError(f"Expected Eval. Observed: {type(item).__name__}") + return selected + + +def _coerce_score(name: str, raw: object) -> tuple[float, str | None, dict[str, Any]]: + if isinstance(raw, Score): + return raw.value, raw.reason, dict(raw.details) + if isinstance(raw, bool): + return (1.0 if raw else 0.0), None, {} + if isinstance(raw, int | float): + validate_unit_interval(f"eval {name} score", raw) + return float(raw), None, {} + raise TypeError( + f"Eval {name!r} must return a Score, float, or bool. Observed: {type(raw).__name__}" + ) diff --git a/src/agent_test_kit/evals/builtin.py b/src/agent_test_kit/evals/builtin.py new file mode 100644 index 0000000..5929954 --- /dev/null +++ b/src/agent_test_kit/evals/builtin.py @@ -0,0 +1,84 @@ +"""Built-in deterministic evals.""" + +from __future__ import annotations + +from collections.abc import Iterable, Sequence + +from agent_test_kit.evals.base import Eval +from agent_test_kit.models.execution import AgentExecutionResult +from agent_test_kit.scoring import RequiredTermsScorer, Score + +ToolSpec = str | tuple[str | None, str] + + +class RequiredTermsEval(Eval): + """Fraction of required terms present in the response text.""" + + def __init__( + self, + terms: Iterable[str], + *, + threshold: float = 1.0, + name: str = "required_terms", + case_sensitive: bool = False, + ) -> None: + super().__init__(name=name, threshold=threshold) + self._scorer = RequiredTermsScorer(terms, case_sensitive=case_sensitive, name=name) + + def score(self, result: AgentExecutionResult) -> Score: + return self._scorer.score(result) + + +class RequiredToolsEval(Eval): + """Fraction of required tools that appear in the trace. + + Each tool is a name, or a ``(server, name)`` pair. Order does not matter. + """ + + def __init__( + self, + tools: Sequence[ToolSpec], + *, + threshold: float = 1.0, + name: str = "required_tools", + ) -> None: + super().__init__(name=name, threshold=threshold) + parsed = tuple(_parse_tool(tool) for tool in tools) + if not parsed: + raise ValueError("at least one tool is required") + self.tools = parsed + + def score(self, result: AgentExecutionResult) -> Score: + missing = [ + _qualified(server, tool_name) + for server, tool_name in self.tools + if not any(call.matches(tool_name, server) for call in result.tool_calls) + ] + found = len(self.tools) - len(missing) + return Score( + found / len(self.tools), + reason=f"Missing tools: {missing}" if missing else None, + details={"missing": missing}, + ) + + +def _parse_tool(tool: ToolSpec) -> tuple[str | None, str]: + if isinstance(tool, str): + if not tool: + raise ValueError("tool names must be non-empty strings") + return None, tool + if ( + isinstance(tool, tuple) + and len(tool) == 2 + and (tool[0] is None or isinstance(tool[0], str)) + and isinstance(tool[1], str) + and tool[1] + ): + return tool + raise ValueError("tools must be names or (server, name) pairs") + + +def _qualified(server: str | None, tool_name: str) -> str: + if server: + return f"{server}/{tool_name}" + return tool_name diff --git a/src/agent_test_kit/guardrails/__init__.py b/src/agent_test_kit/guardrails/__init__.py new file mode 100644 index 0000000..7510032 --- /dev/null +++ b/src/agent_test_kit/guardrails/__init__.py @@ -0,0 +1,33 @@ +"""Offline and online guardrails.""" + +from agent_test_kit.guardrails.base import ( + Guardrail, + GuardrailCheck, + GuardrailEvent, + GuardrailMonitor, + GuardrailReport, + GuardrailViolation, + GuardrailViolationError, + check_guardrails, + guardrail, +) +from agent_test_kit.guardrails.builtin import ( + NoProfanityGuardrail, + RequiredFinalToolGuardrail, + must_open_pull_request, +) + +__all__ = [ + "Guardrail", + "GuardrailCheck", + "GuardrailEvent", + "GuardrailMonitor", + "GuardrailReport", + "GuardrailViolation", + "GuardrailViolationError", + "NoProfanityGuardrail", + "RequiredFinalToolGuardrail", + "check_guardrails", + "guardrail", + "must_open_pull_request", +] diff --git a/src/agent_test_kit/guardrails/base.py b/src/agent_test_kit/guardrails/base.py new file mode 100644 index 0000000..6ca1276 --- /dev/null +++ b/src/agent_test_kit/guardrails/base.py @@ -0,0 +1,489 @@ +"""Guardrails that share one definition across recorded traces and live runs.""" + +from __future__ import annotations + +import inspect +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Literal + +from agent_test_kit.models.execution import AgentExecutionResult +from agent_test_kit.models.tool_call import ToolCall +from agent_test_kit.models.trace import AgentTrace +from agent_test_kit.scoring import response_text + +GuardrailPhase = Literal["message", "tool_call", "trace"] +GuardrailMode = Literal["offline", "online"] + + +@dataclass(frozen=True, slots=True) +class GuardrailViolation: + """One guardrail failure.""" + + guardrail: str + message: str + phase: GuardrailPhase = "trace" + evidence: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class GuardrailCheck: + """Outcome of one guardrail against a trace or a live session.""" + + name: str + passed: bool + mode: GuardrailMode + message: str | None = None + violations: tuple[GuardrailViolation, ...] = () + + def assert_passed(self) -> None: + if not self.passed: + raise AssertionError(f"Guardrail {self.name!r} failed: {self.message}") + + +@dataclass(frozen=True, slots=True) +class GuardrailReport: + """Aggregate of every guardrail checked together.""" + + checks: tuple[GuardrailCheck, ...] + mode: GuardrailMode + + @property + def passed(self) -> bool: + return all(check.passed for check in self.checks) + + @property + def violations(self) -> tuple[GuardrailViolation, ...]: + return tuple(violation for check in self.checks for violation in check.violations) + + def assert_passed(self) -> None: + if self.passed: + return + lines = [f"{check.name}: {check.message}" for check in self.checks if not check.passed] + raise AssertionError("Guardrail violations:\n" + "\n".join(lines)) + + +@dataclass(frozen=True, slots=True) +class GuardrailEvent: + """One streamed message or tool call seen by an online guardrail.""" + + type: Literal["message", "tool_call"] + text: str | None = None + role: str = "assistant" + tool_call: ToolCall | None = None + + +class GuardrailViolationError(AssertionError): + """Raised when an online guardrail is configured to fail on the first violation.""" + + def __init__(self, violation: GuardrailViolation) -> None: + self.violation = violation + super().__init__(f"Guardrail {violation.guardrail!r} failed: {violation.message}") + + +class Guardrail: + """Base class for a guardrail that works offline and online. + + Override the hooks you need. ``check`` replays a recorded result through + those hooks. ``GuardrailMonitor`` calls the same hooks as events arrive. + Return a violation message from a hook, or ``None`` when the event is allowed. + """ + + name: str = "guardrail" + + def on_message(self, text: str, *, role: str = "assistant") -> str | None: + """Inspect one message. Assistant output is the usual place to look.""" + return None + + def on_tool_call(self, call: ToolCall) -> str | None: + """Inspect one tool call as it happens.""" + return None + + def on_trace(self, result: AgentExecutionResult) -> str | None: + """Inspect the finished run. Online monitors call this from ``finish``.""" + return None + + def check(self, result: AgentExecutionResult) -> GuardrailCheck: + """Run this guardrail against a recorded result.""" + violations: list[GuardrailViolation] = [] + text = response_text(result) + if text.strip(): + found = self._message_violation(text, role="assistant") + if found is not None: + violations.append(found) + for call in result.tool_calls: + found = self._tool_violation(call) + if found is not None: + violations.append(found) + found = self._trace_violation(result) + if found is not None: + violations.append(found) + return _check_from_violations(self.name, violations, mode="offline") + + def __call__(self, result: AgentExecutionResult) -> GuardrailCheck: + return self.check(result) + + def _message_violation(self, text: str, *, role: str) -> GuardrailViolation | None: + message = _coerce_violation(self.name, self.on_message(text, role=role)) + if message is None: + return None + return GuardrailViolation( + guardrail=self.name, + message=message, + phase="message", + evidence={"role": role}, + ) + + def _tool_violation(self, call: ToolCall) -> GuardrailViolation | None: + message = _coerce_violation(self.name, self.on_tool_call(call)) + if message is None: + return None + return GuardrailViolation( + guardrail=self.name, + message=message, + phase="tool_call", + evidence={"server": call.server, "tool": call.name}, + ) + + def _trace_violation(self, result: AgentExecutionResult) -> GuardrailViolation | None: + message = _coerce_violation(self.name, self.on_trace(result)) + if message is None: + return None + return GuardrailViolation(guardrail=self.name, message=message, phase="trace") + + +def check_guardrails( + result: AgentExecutionResult, + guardrails: Guardrail | Sequence[Guardrail], +) -> GuardrailReport: + """Run guardrails offline against a recorded execution result.""" + selected = _as_guardrails(guardrails) + checks = tuple(item.check(result) for item in selected) + return GuardrailReport(checks=checks, mode="offline") + + +class GuardrailMonitor: + """Online middleware. Feed messages and tool calls as the agent produces them. + + Message and tool-call hooks run immediately. Trace hooks, including a required + final tool, run when ``finish`` is called. ``fail_fast=True`` raises + ``GuardrailViolationError`` at the first violation so the caller can stop the run. + """ + + def __init__( + self, + guardrails: Guardrail | Sequence[Guardrail], + *, + fail_fast: bool = False, + on_finish: Callable[[GuardrailReport], None] | None = None, + ) -> None: + self.guardrails = _as_guardrails(guardrails) + self.fail_fast = fail_fast + self._on_finish = on_finish + self._violations: list[list[GuardrailViolation]] = [[] for _ in self.guardrails] + self._messages: list[tuple[str, str]] = [] + self._calls: list[ToolCall] = [] + self._finished = False + self.last_report: GuardrailReport | None = None + + @property + def finished(self) -> bool: + return self._finished + + def on_message(self, text: str, *, role: str = "assistant") -> list[GuardrailViolation]: + """Record one streamed message and run message hooks.""" + self._ensure_open() + if not isinstance(text, str): + raise TypeError("message text must be a string") + self._messages.append((role, text)) + found = self._run_message_hooks(text, role=role) + self._raise_fast(found) + return found + + def on_tool_call(self, call: ToolCall | Mapping[str, Any]) -> list[GuardrailViolation]: + """Record one tool call and run tool-call hooks.""" + self._ensure_open() + tool_call = call if isinstance(call, ToolCall) else ToolCall.model_validate(dict(call)) + self._calls.append(tool_call) + found = self._run_tool_hooks(tool_call) + self._raise_fast(found) + return found + + def handle(self, event: GuardrailEvent | Mapping[str, Any]) -> list[GuardrailViolation]: + """Middleware entry point for one streamed event. + + Message events use ``type="message"`` and ``text`` (or ``content``). + Tool events use ``type="tool_call"`` plus ``name`` or a nested ``tool_call``. + """ + parsed = event if isinstance(event, GuardrailEvent) else _parse_event(event) + if parsed.type == "message": + return self.on_message(parsed.text or "", role=parsed.role) + if parsed.tool_call is None: + raise ValueError("tool_call event requires a tool call") + return self.on_tool_call(parsed.tool_call) + + def finish( + self, + result: AgentExecutionResult | None = None, + *, + notify: bool = True, + raise_on_violation: bool | None = None, + ) -> GuardrailReport: + """Finish the live run and evaluate trace-level guardrails. + + When ``result`` is omitted, the trace is the messages and tool calls already + observed. ``raise_on_violation`` defaults to ``fail_fast``. + """ + if self._finished: + raise RuntimeError("GuardrailMonitor.finish() was already called") + should_raise = self.fail_fast if raise_on_violation is None else raise_on_violation + self._replay(result) + self._scan_joined_messages() + trace = result if result is not None else self._synthetic_result() + self._run_trace_hooks(trace) + self._finished = True + report = self._build_report() + self.last_report = report + if result is not None: + result.guardrail_checks.extend(report.checks) + if should_raise and not report.passed: + raise GuardrailViolationError(report.violations[0]) + if notify and self._on_finish is not None: + self._on_finish(report) + return report + + def _ensure_open(self) -> None: + if self._finished: + raise RuntimeError("GuardrailMonitor is already finished") + + def _raise_fast(self, found: list[GuardrailViolation]) -> None: + if self.fail_fast and found: + self.last_report = self._build_report() + raise GuardrailViolationError(found[0]) + + def _scan_joined_messages(self) -> None: + """Catch a disallowed word that was split across streamed chunks.""" + assistant = [message for role, message in self._messages if role == "assistant"] + joined = "".join(assistant) + if joined.strip() and joined not in assistant: + self._run_message_hooks(joined, role="assistant") + + def _replay(self, result: AgentExecutionResult | None) -> None: + if result is None: + return + text = response_text(result) + observed = [message for role, message in self._messages if role == "assistant"] + if text.strip() and text not in observed and "".join(observed) != text: + self._messages.append(("assistant", text)) + self._run_message_hooks(text, role="assistant") + if not self._calls: + for call in result.tool_calls: + self._calls.append(call) + self._run_tool_hooks(call) + + def _run_message_hooks(self, text: str, *, role: str) -> list[GuardrailViolation]: + found: list[GuardrailViolation] = [] + for index, guard in enumerate(self.guardrails): + violation = guard._message_violation(text, role=role) + if violation is not None: + self._violations[index].append(violation) + found.append(violation) + return found + + def _run_tool_hooks(self, call: ToolCall) -> list[GuardrailViolation]: + found: list[GuardrailViolation] = [] + for index, guard in enumerate(self.guardrails): + violation = guard._tool_violation(call) + if violation is not None: + self._violations[index].append(violation) + found.append(violation) + return found + + def _run_trace_hooks(self, result: AgentExecutionResult) -> None: + for index, guard in enumerate(self.guardrails): + violation = guard._trace_violation(result) + if violation is not None: + self._violations[index].append(violation) + + def _synthetic_result(self) -> AgentExecutionResult: + text = "".join(message for role, message in self._messages if role == "assistant") + return AgentExecutionResult( + success=True, + run_id="online", + response=text or None, + trace=AgentTrace(tool_calls=list(self._calls)), + ) + + def _build_report(self) -> GuardrailReport: + checks = tuple( + _check_from_violations(guard.name, self._violations[index], mode="online") + for index, guard in enumerate(self.guardrails) + ) + return GuardrailReport(checks=checks, mode="online") + + +def guardrail( + func: Callable[..., str | None] | None = None, + *, + name: str | None = None, + phase: GuardrailPhase = "trace", +) -> Any: + """Turn a function into a guardrail. + + ``phase="message"`` receives ``(text, *, role)``. ``role`` is omitted when the + function does not accept it. ``phase="tool_call"`` receives a ``ToolCall``. + ``phase="trace"`` receives an ``AgentExecutionResult``. Return a violation + message, or ``None`` when the event is allowed. + """ + if phase not in {"message", "tool_call", "trace"}: + raise ValueError("phase must be 'message', 'tool_call', or 'trace'") + + def decorate(fn: Callable[..., str | None]) -> Guardrail: + label = _require_name(name or fn.__name__, what="guardrail name") + + class _FunctionGuardrail(Guardrail): + pass + + _FunctionGuardrail.name = label + _FunctionGuardrail.__name__ = label + _FunctionGuardrail.__doc__ = fn.__doc__ + if phase == "message": + accepts_role = _accepts_keyword(fn, "role") + + def on_message( + self: Guardrail, + text: str, + *, + role: str = "assistant", + ) -> str | None: + if accepts_role: + return fn(text, role=role) + return fn(text) + + _FunctionGuardrail.on_message = on_message # type: ignore[method-assign] + elif phase == "tool_call": + + def on_tool_call(self: Guardrail, call: ToolCall) -> str | None: + return fn(call) + + _FunctionGuardrail.on_tool_call = on_tool_call # type: ignore[method-assign] + else: + + def on_trace(self: Guardrail, result: AgentExecutionResult) -> str | None: + return fn(result) + + _FunctionGuardrail.on_trace = on_trace # type: ignore[method-assign] + return _FunctionGuardrail() + + if func is not None: + return decorate(func) + return decorate + + +def _as_guardrails(guardrails: Guardrail | Sequence[Guardrail]) -> tuple[Guardrail, ...]: + selected = (guardrails,) if isinstance(guardrails, Guardrail) else tuple(guardrails) + if not selected: + raise ValueError("at least one guardrail is required") + for item in selected: + if not isinstance(item, Guardrail): + raise TypeError(f"Expected Guardrail. Observed: {type(item).__name__}") + return selected + + +def _coerce_violation(name: str, value: object) -> str | None: + if value is None: + return None + if isinstance(value, str): + text = value.strip() + return text or None + raise TypeError( + f"Guardrail {name!r} must return a violation message or None. " + f"Observed: {type(value).__name__}" + ) + + +def _check_from_violations( + name: str, + violations: Sequence[GuardrailViolation], + *, + mode: GuardrailMode, +) -> GuardrailCheck: + unique = _dedupe_violations(violations) + if not unique: + return GuardrailCheck(name=name, passed=True, mode=mode) + message = "; ".join(item.message for item in unique) + return GuardrailCheck( + name=name, + passed=False, + mode=mode, + message=message, + violations=unique, + ) + + +def _dedupe_violations( + violations: Sequence[GuardrailViolation], +) -> tuple[GuardrailViolation, ...]: + unique: list[GuardrailViolation] = [] + seen: set[tuple[str, str, str]] = set() + for violation in violations: + key = (violation.guardrail, violation.phase, violation.message) + if key in seen: + continue + seen.add(key) + unique.append(violation) + return tuple(unique) + + +def _require_name(name: str, *, what: str) -> str: + if not isinstance(name, str) or not name.strip(): + raise ValueError(f"{what} must be a non-empty string") + return name + + +def _accepts_keyword(func: Callable[..., Any], keyword: str) -> bool: + try: + signature = inspect.signature(func) + except (TypeError, ValueError): + return True + for parameter in signature.parameters.values(): + if parameter.kind == inspect.Parameter.VAR_KEYWORD: + return True + if parameter.name == keyword: + return True + return False + + +def _parse_event(event: Mapping[str, Any]) -> GuardrailEvent: + if isinstance(event, str) or not isinstance(event, Mapping): + raise TypeError("guardrail event must be a mapping or GuardrailEvent") + event_type = event.get("type") + if event_type == "message": + text = event.get("text", event.get("content", "")) + if text is None: + text = "" + if not isinstance(text, str): + raise TypeError("message text must be a string") + role = event.get("role") or "assistant" + if not isinstance(role, str): + raise TypeError("message role must be a string") + return GuardrailEvent(type="message", text=text, role=role) + if event_type == "tool_call": + return GuardrailEvent(type="tool_call", tool_call=_tool_call_from_event(event)) + raise ValueError(f"Unknown guardrail event type: {event_type!r}") + + +def _tool_call_from_event(payload: Mapping[str, Any]) -> ToolCall: + nested = payload.get("tool_call", payload.get("tool")) + if isinstance(nested, ToolCall): + return nested + if isinstance(nested, Mapping): + return ToolCall.model_validate(dict(nested)) + data = { + key: payload[key] + for key in ("name", "server", "arguments", "status", "operation_kind", "output", "id") + if key in payload + } + if "name" not in data: + raise ValueError("tool_call event requires a name") + return ToolCall.model_validate(data) diff --git a/src/agent_test_kit/guardrails/builtin.py b/src/agent_test_kit/guardrails/builtin.py new file mode 100644 index 0000000..6184187 --- /dev/null +++ b/src/agent_test_kit/guardrails/builtin.py @@ -0,0 +1,140 @@ +"""Built-in guardrails for assistant language and required final tool calls.""" + +from __future__ import annotations + +import re +from collections.abc import Iterable + +from agent_test_kit.guardrails.base import Guardrail +from agent_test_kit.models.execution import AgentExecutionResult + +_DEFAULT_PROFANITY: tuple[str, ...] = ( + "fuck", + "fucking", + "fucked", + "shit", + "shitty", + "bitch", + "asshole", + "bastard", + "dick", + "piss", +) + + +class NoProfanityGuardrail(Guardrail): + """Reject disallowed language in assistant output. + + The default term list is a small profanity set. Pass ``terms`` to replace it + with any disallowed language, or ``extra_terms`` to add phrases. Matching is + case-insensitive. A single word uses word boundaries; a phrase matches as text. + """ + + def __init__( + self, + terms: Iterable[str] | None = None, + *, + extra_terms: Iterable[str] = (), + name: str = "no_profanity", + ) -> None: + selected = tuple(terms) if terms is not None else _DEFAULT_PROFANITY + combined = _validated_terms((*selected, *extra_terms)) + self.terms = combined + self._compiled = tuple(_compile_term(term) for term in combined) + self.name = name + + def on_message(self, text: str, *, role: str = "assistant") -> str | None: + if role != "assistant" or not text: + return None + found = _find_terms(text, self.terms, self._compiled) + if not found: + return None + return f"Disallowed language in assistant output: {', '.join(found)}" + + +class RequiredFinalToolGuardrail(Guardrail): + """The run must finish with a successful call to ``tool_name``. + + Use this for flow rules such as "this agent must open a pull request at the end". + The check runs on the finished trace, offline or when an online monitor finishes. + """ + + def __init__( + self, + tool_name: str, + *, + server: str | None = None, + name: str | None = None, + require_success: bool = True, + ) -> None: + if not isinstance(tool_name, str) or not tool_name: + raise ValueError("tool_name must be a non-empty string") + self.tool_name = tool_name + self.server = server + self.require_success = require_success + self.name = name or "required_final_tool" + + def on_trace(self, result: AgentExecutionResult) -> str | None: + expected = _qualified(self.server, self.tool_name) + calls = result.tool_calls + if not calls: + return f"Expected final tool {expected}; the run recorded no tool calls" + last = calls[-1] + if not last.matches(self.tool_name, self.server): + return f"Expected final tool {expected}; observed {_qualified(last.server, last.name)}" + if self.require_success and not last.is_success: + status = last.status.value if hasattr(last.status, "value") else str(last.status) + return f"Final tool {expected} finished with status {status}" + return None + + +def must_open_pull_request( + tool_name: str = "create_pull_request", + *, + server: str | None = None, +) -> RequiredFinalToolGuardrail: + """Flow rule: the agent must open a pull request as its final tool call.""" + return RequiredFinalToolGuardrail( + tool_name, + server=server, + name="must_open_pull_request", + ) + + +def _validated_terms(terms: tuple[str, ...]) -> tuple[str, ...]: + unique: list[str] = [] + for term in terms: + if not isinstance(term, str) or not term.strip(): + raise ValueError("disallowed terms must be non-empty strings") + if term not in unique: + unique.append(term) + if not unique: + raise ValueError("at least one disallowed term is required") + return tuple(unique) + + +def _compile_term(term: str) -> tuple[str, re.Pattern[str] | None]: + needle = term.casefold() + if any(character.isspace() for character in needle) or not needle.isalnum(): + return needle, None + return needle, re.compile(rf"\b{re.escape(needle)}\b") + + +def _find_terms( + text: str, + terms: tuple[str, ...], + compiled: tuple[tuple[str, re.Pattern[str] | None], ...], +) -> list[str]: + folded = text.casefold() + found: list[str] = [] + for original, (needle, pattern) in zip(terms, compiled, strict=True): + matched = needle in folded if pattern is None else pattern.search(folded) is not None + if matched: + found.append(original) + return found + + +def _qualified(server: str | None, tool_name: str) -> str: + if server: + return f"{server}/{tool_name}" + return tool_name diff --git a/src/agent_test_kit/models/execution.py b/src/agent_test_kit/models/execution.py index 875d0c1..fd5b4d8 100644 --- a/src/agent_test_kit/models/execution.py +++ b/src/agent_test_kit/models/execution.py @@ -18,6 +18,8 @@ StatusFilter, WorkflowStep, ) + from agent_test_kit.evals.base import Eval, EvalReport + from agent_test_kit.guardrails.base import Guardrail, GuardrailReport from agent_test_kit.models.enums import ToolCallStatus from agent_test_kit.scoring import Score, Scorer @@ -47,6 +49,8 @@ class AgentExecutionResult(BaseModel): duration_ms: int | None = None assertions: list[str] = Field(default_factory=list) assertion_outcomes: list[AssertionOutcome] = Field(default_factory=list) + guardrail_checks: list[Any] = Field(default_factory=list, exclude=True) + eval_checks: list[Any] = Field(default_factory=list, exclude=True) model_config = {"arbitrary_types_allowed": True} @@ -359,6 +363,48 @@ def evaluate() -> None: ) return score + def assert_guardrails(self, guardrails: Sequence[Guardrail] | Guardrail) -> GuardrailReport: + """Run guardrails offline and record the outcome like other assertions.""" + from agent_test_kit.guardrails.base import check_guardrails + + report = check_guardrails(self, guardrails) + self.guardrail_checks.extend(report.checks) + + def evaluate() -> None: + report.assert_passed() + + self._capture_assertion( + "assert_guardrails", + expected=[item.name for item in report.checks], + actual=[ + {"name": check.name, "passed": check.passed, "message": check.message} + for check in report.checks + ], + assertion=evaluate, + ) + return report + + def assert_evals(self, evals: Sequence[Eval] | Eval) -> EvalReport: + """Score evals and record pass/fail like other assertions.""" + from agent_test_kit.evals.base import run_evals + + report = run_evals(self, evals) + self.eval_checks.extend(report.results) + + def evaluate() -> None: + report.assert_passed() + + self._capture_assertion( + "assert_evals", + expected=[{"name": item.name, "threshold": item.threshold} for item in report.results], + actual=[ + {"name": item.name, "score": item.score, "passed": item.passed} + for item in report.results + ], + assertion=evaluate, + ) + return report + @classmethod def from_api_response( cls, diff --git a/src/agent_test_kit/models/report.py b/src/agent_test_kit/models/report.py index 688f0a4..4e2984c 100644 --- a/src/agent_test_kit/models/report.py +++ b/src/agent_test_kit/models/report.py @@ -26,6 +26,36 @@ class AssertionReport(BaseModel): message: str | None = None +class GuardrailViolationReport(BaseModel): + """One guardrail violation included in a scenario report.""" + + guardrail: str + message: str + phase: Literal["message", "tool_call", "trace"] = "trace" + evidence: dict[str, Any] = Field(default_factory=dict) + + +class GuardrailCheckReport(BaseModel): + """Outcome of one guardrail, offline or online.""" + + name: str + passed: bool + mode: Literal["offline", "online"] = "offline" + message: str | None = None + violations: list[GuardrailViolationReport] = Field(default_factory=list) + + +class EvalCheckReport(BaseModel): + """Score and pass/fail for one eval.""" + + name: str + passed: bool + score: float + threshold: float + reason: str | None = None + details: dict[str, Any] = Field(default_factory=dict) + + class SideEffectVerificationReport(BaseModel): """Outcome of an external side-effect verifier.""" @@ -53,6 +83,8 @@ class ScenarioReport(BaseModel): tool_calls: list[ToolCall] = Field(default_factory=list) timeline: list[TraceEvent] = Field(default_factory=list) assertions: list[AssertionReport] = Field(default_factory=list) + guardrails: list[GuardrailCheckReport] = Field(default_factory=list) + evals: list[EvalCheckReport] = Field(default_factory=list) side_effect_verifications: list[SideEffectVerificationReport] = Field(default_factory=list) error_message: str | None = None retry_count: int = 0 @@ -65,6 +97,7 @@ class ScenarioReport(BaseModel): suggested_prompt: str | None = None endpoint_base_url: str | None = None connected_mcp_servers: list[str] = Field(default_factory=list) + expected_mcp_servers: list[str] = Field(default_factory=list) injection_detected: bool | None = None violation_types: list[str] = Field(default_factory=list) golden_outcome: str | None = None diff --git a/src/agent_test_kit/plugin/pytest_plugin.py b/src/agent_test_kit/plugin/pytest_plugin.py index 4a400d0..449b557 100644 --- a/src/agent_test_kit/plugin/pytest_plugin.py +++ b/src/agent_test_kit/plugin/pytest_plugin.py @@ -3,7 +3,7 @@ from __future__ import annotations import os -from collections.abc import AsyncIterator, Sequence +from collections.abc import AsyncIterator, Iterator, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any @@ -16,6 +16,13 @@ from agent_test_kit.client.config import AgentTestConfig from agent_test_kit.client.cursor_agent_client import CursorAgentClient from agent_test_kit.client.prompt_ai_helper_profiles import get_profile +from agent_test_kit.evals.base import Eval, EvalReport, get_eval +from agent_test_kit.guardrails.base import ( + Guardrail, + GuardrailMonitor, + GuardrailReport, + check_guardrails, +) from agent_test_kit.models.execution import AgentExecutionResult from agent_test_kit.models.report import AgentTestRunReport, ReleaseReadiness, ScenarioStatus from agent_test_kit.reporting.html_report import HtmlReportWriter @@ -40,6 +47,8 @@ class PluginState: results: dict[str, AgentExecutionResult] = field(default_factory=dict) verification_results: dict[str, list[VerificationResult]] = field(default_factory=dict) regression_evaluations: dict[str, RegressionRunEvaluation] = field(default_factory=dict) + guardrail_checks: dict[str, list[Any]] = field(default_factory=dict) + eval_checks: dict[str, list[Any]] = field(default_factory=dict) phase_reports: dict[str, dict[str, pytest.TestReport]] = field(default_factory=dict) html_path: Path | None = None write_json: bool = True @@ -121,6 +130,14 @@ def attach_regression_evaluation(self, evaluation: RegressionRunEvaluation) -> N if state is not None: state.regression_evaluations[self.nodeid] = evaluation + def attach_guardrail_report(self, report: GuardrailReport) -> None: + """Record guardrail checks on the current scenario.""" + attach_guardrail_report(self.config, self.nodeid, report) + + def attach_eval_report(self, report: EvalReport) -> None: + """Record eval scores on the current scenario.""" + attach_eval_report(self.config, self.nodeid, report) + async def verify( self, verifiers: Sequence[SideEffectVerifier], @@ -187,6 +204,106 @@ def agent_scenario(request: pytest.FixtureRequest) -> ScenarioResultRecorder: return ScenarioResultRecorder(request.config, request.node.nodeid) +@dataclass +class _GuardrailFixture: + """Run guardrails for the current test and attach them to the report.""" + + config: pytest.Config + nodeid: str + monitors: list[GuardrailMonitor] = field(default_factory=list) + _closed: bool = False + + def check( + self, + result: AgentExecutionResult, + guardrails: Guardrail | Sequence[Guardrail], + *, + fail: bool = True, + ) -> GuardrailReport: + """Run guardrails offline. Failures are recorded even when the assertion fails.""" + report = check_guardrails(result, guardrails) + result.guardrail_checks.extend(report.checks) + attach_guardrail_report(self.config, self.nodeid, report) + if fail: + report.assert_passed() + return report + + def monitor( + self, + guardrails: Guardrail | Sequence[Guardrail], + *, + fail_fast: bool = False, + ) -> GuardrailMonitor: + """Online hook. Trace rules run when the monitor or this fixture finishes.""" + monitor = GuardrailMonitor(guardrails, fail_fast=fail_fast) + self.monitors.append(monitor) + return monitor + + def close(self) -> None: + if self._closed: + return + self._closed = True + for monitor in self.monitors: + if not monitor.finished: + monitor.finish(notify=False, raise_on_violation=False) + if monitor.last_report is not None: + attach_guardrail_report(self.config, self.nodeid, monitor.last_report) + + +@dataclass +class _EvalFixture: + """Run registered or explicit evals for the current test.""" + + config: pytest.Config + nodeid: str + names: list[str] + + def run( + self, + result: AgentExecutionResult, + evals: Eval | Sequence[Eval] | None = None, + *, + fail: bool = True, + ) -> EvalReport: + """Score evals. Marker names are used when ``evals`` is omitted.""" + from agent_test_kit.evals.base import run_evals + + selected: Eval | Sequence[Eval] + if evals is None: + if not self.names: + raise RuntimeError( + "No evals to run. Pass evals=[...] or mark the test with " + "@pytest.mark.agent_eval('registered_name')." + ) + selected = [get_eval(name) for name in self.names] + else: + selected = evals + report = run_evals(result, selected) + result.eval_checks.extend(report.results) + attach_eval_report(self.config, self.nodeid, report) + if fail: + report.assert_passed() + return report + + +@pytest.fixture +def agent_guardrails(request: pytest.FixtureRequest) -> Iterator[_GuardrailFixture]: + """Offline ``check`` and online ``monitor`` for guardrails on this test.""" + runner = _GuardrailFixture(request.config, request.node.nodeid) + yield runner + runner.close() + + +@pytest.fixture +def agent_eval(request: pytest.FixtureRequest) -> _EvalFixture: + """Run evals named by ``@pytest.mark.agent_eval`` or passed to ``run``.""" + return _EvalFixture( + request.config, + request.node.nodeid, + _agent_eval_names(request.node), + ) + + @pytest_asyncio.fixture async def agent_cleanup() -> AsyncIterator[CleanupManager]: manager = CleanupManager() @@ -245,6 +362,8 @@ def _record_scenario_from_reports( execution_result=stored_result, verification_results=state.verification_results.get(item.nodeid, []), regression_evaluation=state.regression_evaluations.get(item.nodeid), + guardrail_checks=state.guardrail_checks.get(item.nodeid, []), + eval_checks=state.eval_checks.get(item.nodeid, []), ) @@ -316,6 +435,46 @@ def store_execution_result( state.results[nodeid] = result +def _agent_eval_names(node: pytest.Item | pytest.Collector) -> list[str]: + names: list[str] = [] + for marker in node.iter_markers("agent_eval"): + for arg in marker.args: + if not isinstance(arg, str) or not arg.strip(): + raise ValueError("agent_eval marker arguments must be non-empty eval names") + names.append(arg) + return names + + +def attach_guardrail_report( + config: pytest.Config, + nodeid: str, + report: GuardrailReport, +) -> None: + """Attach guardrail checks to a report scenario.""" + state = config.stash.get(_STASH_KEY, None) + if state is None: + return + bucket = state.guardrail_checks.setdefault(nodeid, []) + for check in report.checks: + if check not in bucket: + bucket.append(check) + + +def attach_eval_report( + config: pytest.Config, + nodeid: str, + report: EvalReport, +) -> None: + """Attach eval results to a report scenario.""" + state = config.stash.get(_STASH_KEY, None) + if state is None: + return + bucket = state.eval_checks.setdefault(nodeid, []) + for result in report.results: + if result not in bucket: + bucket.append(result) + + def attach_verification_results( config: pytest.Config, nodeid: str, diff --git a/src/agent_test_kit/reporting/html_report.py b/src/agent_test_kit/reporting/html_report.py index 592c4e6..c77e431 100644 --- a/src/agent_test_kit/reporting/html_report.py +++ b/src/agent_test_kit/reporting/html_report.py @@ -72,7 +72,7 @@ def _flow_sequence(tool_calls: list[dict[str, Any]]) -> str: step = ( f'' f'{escape(kind.upper())}' - f"{escape(label)}" + f'{escape(label)}' f"" ) steps.append(step) @@ -217,6 +217,24 @@ def _readiness_block(readiness: dict[str, Any] | None) -> str: return "".join(parts) +def _phases(item: dict[str, Any]) -> str: + phases: list[str] = [] + for violation in item.get("violations") or []: + phase = str(violation.get("phase") or "") + if phase and phase not in phases: + phases.append(phase) + return ", ".join(phases) if phases else "—" + + +def _optional_details(title: str, headers: list[str], rows: list[list[Any]]) -> str: + if not rows: + return "" + return ( + f"

{escape(title)} ({len(rows)})" + f"{_table(headers, rows)}
" + ) + + def _behavior_rows(scenario: dict[str, Any]) -> list[list[Any]]: rows: list[list[Any]] = [] if scenario.get("case_id"): @@ -384,6 +402,26 @@ def _scenario(self, scenario: dict[str, Any]) -> str: ] for assertion in scenario["assertions"] ] + guardrail_rows = [ + [ + item["name"], + item.get("mode"), + item["passed"], + _phases(item), + item.get("message"), + ] + for item in scenario.get("guardrails") or [] + ] + eval_rows = [ + [ + item["name"], + f"{float(item['score']):.2f}", + f"{float(item['threshold']):.2f}", + item["passed"], + item.get("reason"), + ] + for item in scenario.get("evals") or [] + ] verification_rows = [ [ verification["verifier_name"], @@ -423,9 +461,7 @@ def _scenario(self, scenario: dict[str, Any]) -> str: ) flow_html = _flow_sequence(scenario["tool_calls"]) kind = scenario.get("scenario_kind") - kind_badge = ( - f'{escape(str(kind))}' if kind else "" - ) + kind_badge = f'{escape(str(kind))}' if kind else "" prompt_review_html = _prompt_review_block(scenario) golden_html = _golden_outcome_block(scenario) connected = scenario.get("connected_mcp_servers") or [] @@ -442,7 +478,9 @@ def _scenario(self, scenario: dict[str, Any]) -> str: injection_html = f'{label}' status_class = escape(status) return f"""
-

{_text(scenario["name"])}{kind_badge}{injection_html} — {escape(status)}

+

{_text(scenario["name"])}{kind_badge}{injection_html} — { + escape(status) + }

{_table(["Field", "Value"], summary)}

Connected MCP servers

{mcp_html} @@ -452,6 +490,20 @@ def _scenario(self, scenario: dict[str, Any]) -> str: {flow_html}
Assertions ({len(assertion_rows)}) {_table(["Name", "Passed", "Expected", "Actual", "Message"], assertion_rows)}
+{ + _optional_details( + "Guardrails", + ["Guardrail", "Mode", "Passed", "Phase", "Message"], + guardrail_rows, + ) + } +{ + _optional_details( + "Evals", + ["Eval", "Score", "Threshold", "Passed", "Reason"], + eval_rows, + ) + }
Side effects ({len(verification_rows)}) {verification_table}
Tool / MCP calls ({len(tool_rows)}) diff --git a/src/agent_test_kit/reporting/json_report.py b/src/agent_test_kit/reporting/json_report.py index a508e88..6fc16ab 100644 --- a/src/agent_test_kit/reporting/json_report.py +++ b/src/agent_test_kit/reporting/json_report.py @@ -14,6 +14,9 @@ from agent_test_kit.models.report import ( AgentTestRunReport, AssertionReport, + EvalCheckReport, + GuardrailCheckReport, + GuardrailViolationReport, ScenarioReport, ScenarioStatus, SideEffectVerificationReport, @@ -26,6 +29,51 @@ from agent_test_kit.regression import RegressionRunEvaluation +def _unique(items: Sequence[Any]) -> list[Any]: + unique: list[Any] = [] + for item in items: + if item not in unique: + unique.append(item) + return unique + + +def _guardrail_reports(checks: Sequence[Any]) -> list[GuardrailCheckReport]: + reports: list[GuardrailCheckReport] = [] + for check in checks: + reports.append( + GuardrailCheckReport( + name=check.name, + passed=bool(check.passed), + mode=check.mode, + message=check.message, + violations=[ + GuardrailViolationReport( + guardrail=item.guardrail, + message=item.message, + phase=item.phase, + evidence=dict(item.evidence), + ) + for item in check.violations + ], + ) + ) + return reports + + +def _eval_reports(checks: Sequence[Any]) -> list[EvalCheckReport]: + return [ + EvalCheckReport( + name=check.name, + passed=bool(check.passed), + score=float(check.score), + threshold=float(check.threshold), + reason=check.reason, + details=dict(check.details), + ) + for check in checks + ] + + def _evidence_links(value: Any) -> list[str]: if isinstance(value, str): return [value] if value.startswith(("https://", "http://")) else [] @@ -68,6 +116,8 @@ def record_scenario( execution_result: AgentExecutionResult | None = None, verification_results: Sequence[VerificationResult] = (), regression_evaluation: RegressionRunEvaluation | None = None, + guardrail_checks: Sequence[Any] = (), + eval_checks: Sequence[Any] = (), ) -> None: extra = extra or {} evaluation = regression_evaluation @@ -88,6 +138,22 @@ def record_scenario( if execution_result is not None else extra.get("assertions", []) ) + guardrail_reports = _guardrail_reports( + _unique( + [ + *(execution_result.guardrail_checks if execution_result is not None else []), + *guardrail_checks, + ] + ) + ) + eval_reports = _eval_reports( + _unique( + [ + *(execution_result.eval_checks if execution_result is not None else []), + *eval_checks, + ] + ) + ) verification_reports = [ SideEffectVerificationReport( verifier_name=result.verifier_name or "unnamed-verifier", @@ -133,6 +199,8 @@ def record_scenario( tool_calls=tool_calls, timeline=execution_result.trace.events if execution_result is not None else [], assertions=assertions, + guardrails=guardrail_reports, + evals=eval_reports, side_effect_verifications=verification_reports, error_message=error_message, retry_count=sum(max(call.attempt - 1, 0) for call in tool_calls), @@ -156,8 +224,7 @@ def record_scenario( or ("prompt_review" if response_dict.get("suggested_prompt") else None) ), original_prompt=( - extra.get("original_prompt") - or response_dict.get("original_prompt") + extra.get("original_prompt") or response_dict.get("original_prompt") ), suggested_prompt=( extra.get("suggested_prompt") @@ -184,9 +251,7 @@ def record_scenario( extra.get("expected_mcp_servers") or self.expected_mcp_servers or [] ), violation_types=list( - extra.get("violation_types") - or response_dict.get("violation_types") - or [] + extra.get("violation_types") or response_dict.get("violation_types") or [] ), golden_outcome=extra.get("golden_outcome") or response_dict.get("golden_outcome"), expected_classification=extra.get("expected_classification") diff --git a/src/agent_test_kit_pytest_entrypoint.py b/src/agent_test_kit_pytest_entrypoint.py index 9a8963e..2709535 100644 --- a/src/agent_test_kit_pytest_entrypoint.py +++ b/src/agent_test_kit_pytest_entrypoint.py @@ -41,6 +41,14 @@ def pytest_addoption(parser: pytest.Parser) -> None: ) +def pytest_configure(config: pytest.Config) -> None: + """Register markers before collection without importing the measured package.""" + config.addinivalue_line( + "markers", + "agent_eval(name): run registered evals by name via the agent_eval fixture", + ) + + @pytest.hookimpl(trylast=True) def pytest_sessionstart(session: pytest.Session) -> None: """Load runtime hooks only after coverage startup is complete.""" diff --git a/tests/test_evals.py b/tests/test_evals.py new file mode 100644 index 0000000..ae5b3d9 --- /dev/null +++ b/tests/test_evals.py @@ -0,0 +1,229 @@ +"""Evals: registration, scoring, and known-good / known-bad traces.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + Eval, + RequiredTermsEval, + RequiredToolsEval, + Score, + ToolCall, + eval_case, + get_eval, + register_eval, + response_text, + run_evals, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace + +pytestmark = pytest.mark.agent_unit + + +def _result( + response: str = "", + calls: list[ToolCall] | None = None, +) -> AgentExecutionResult: + return AgentExecutionResult( + success=True, + run_id="run-eval", + response=response, + trace=AgentTrace(tool_calls=list(calls or [])), + ) + + +def _call(name: str, *, server: str | None = None) -> ToolCall: + return ToolCall( + server=server, + name=name, + status=ToolCallStatus.SUCCESS, + operation_kind=ToolOperationKind.READ, + ) + + +@pytest.mark.parametrize( + ("response", "expect_pass"), + [ + ("Your refund for order 123 is pending.", True), + ("I don't know.", False), + ("The refund is pending.", False), + ], +) +def test_required_terms_eval_known_traces(response: str, expect_pass: bool) -> None: + evaluation = RequiredTermsEval(["refund", "123"]) + result = evaluation.evaluate(_result(response)) + assert result.passed is expect_pass + assert result.name == "required_terms" + if expect_pass: + assert result.score == 1.0 + result.assert_passed() + else: + assert result.score < 1.0 + with pytest.raises(AssertionError, match="required_terms"): + result.assert_passed() + + +@pytest.mark.parametrize( + ("calls", "expect_pass"), + [ + ( + [_call("get_invoice", server="billing"), _call("create_issue", server="jira")], + True, + ), + ([_call("get_invoice", server="billing")], False), + ([], False), + ], +) +def test_required_tools_eval_known_traces(calls: list[ToolCall], expect_pass: bool) -> None: + evaluation = RequiredToolsEval([("billing", "get_invoice"), ("jira", "create_issue")]) + result = evaluation.evaluate(_result("done", calls)) + assert result.passed is expect_pass + if expect_pass: + assert result.score == 1.0 + result.assert_passed() + else: + assert result.score < 1.0 + assert "missing" in result.details + with pytest.raises(AssertionError, match="required_tools"): + result.assert_passed() + + +def test_required_terms_threshold_boundary() -> None: + evaluation = RequiredTermsEval(["refund", "123"], threshold=0.5, name="refund_partial") + partial = evaluation.evaluate(_result("The refund is pending.")) + assert partial.score == 0.5 + partial.assert_passed() + missed = evaluation.evaluate(_result("nothing")) + assert missed.score == 0.0 + with pytest.raises(AssertionError): + missed.assert_passed() + + +def test_assert_evals_records_failure_and_success() -> None: + bad = _result("nope") + with pytest.raises(AssertionError, match="required_terms"): + bad.assert_evals(RequiredTermsEval(["refund"])) + assert bad.assertion_outcomes[-1].name == "assert_evals" + assert bad.assertion_outcomes[-1].passed is False + assert bad.eval_checks[0].passed is False + + mixed = _result("refund ready") + with pytest.raises(AssertionError, match="required_tools"): + mixed.assert_evals([RequiredTermsEval(["refund"]), RequiredToolsEval(["lookup"])]) + assert mixed.eval_checks[0].passed is True + assert mixed.eval_checks[1].passed is False + + passed = _result("refund ready", [_call("lookup")]) + report = passed.assert_evals([RequiredTermsEval(["refund"]), RequiredToolsEval(["lookup"])]) + assert report.passed + assert passed.assertion_outcomes[-1].passed is True + + +def test_eval_case_registers_and_runs() -> None: + @eval_case(name="kit_mentions_ready", threshold=1.0) + def mentions_ready(result: AgentExecutionResult) -> float: + return 1.0 if "ready" in response_text(result).casefold() else 0.0 + + assert get_eval("kit_mentions_ready") is mentions_ready + run_evals(_result("system ready"), mentions_ready).assert_passed() + bad = run_evals(_result("not yet"), mentions_ready) + assert not bad.passed + with pytest.raises(AssertionError, match="kit_mentions_ready"): + bad.assert_passed() + + +def test_eval_case_bool_and_score_returns() -> None: + from agent_test_kit import registered_evals + + @eval_case(name="kit_bool_eval", register=True) + def as_bool(result: AgentExecutionResult) -> bool: + return bool(response_text(result)) + + @eval_case(name="kit_score_eval", threshold=0.8, register=False) + def as_score(result: AgentExecutionResult) -> Score: + return Score(0.8, reason="judge", details={"source": "test"}) + + assert as_bool.evaluate(_result("")).score == 0.0 + assert as_bool.evaluate(_result("hi")).passed is True + scored = as_score.evaluate(_result("hi")) + assert scored.passed is True + assert scored.reason == "judge" + assert scored.details == {"source": "test"} + with pytest.raises(ValueError, match="already registered"): + eval_case(name="kit_bool_eval")(lambda result: 1.0) + register_eval(as_bool) + assert "kit_bool_eval" in registered_evals() + assert "kit_score_eval" not in registered_evals() + + +def test_eval_rejects_bad_scores_and_empty_input() -> None: + class BadEval(Eval): + name = "kit_bad_eval" + + def score(self, result: AgentExecutionResult) -> Any: + return "high" + + class AsyncEval(Eval): + name = "kit_async_eval" + + def score(self, result: AgentExecutionResult) -> Any: + async def inner() -> float: + return 1.0 + + return inner() + + with pytest.raises(TypeError, match="kit_bad_eval"): + BadEval().evaluate(_result("x")) + with pytest.raises(TypeError, match="awaitable"): + AsyncEval().evaluate(_result("x")) + with pytest.raises(ValueError, match="between"): + RequiredTermsEval(["ok"], threshold=1.5) + with pytest.raises(ValueError, match="at least one"): + run_evals(_result("x"), []) + with pytest.raises(ValueError, match="at least one"): + RequiredToolsEval([]) + with pytest.raises(ValueError, match="names or"): + RequiredToolsEval([("only-server",)]) # type: ignore[list-item] + with pytest.raises(KeyError, match="not_a_real_eval"): + get_eval("not_a_real_eval") + with pytest.raises(TypeError, match="Expected Eval"): + run_evals(_result("x"), ["nope"]) # type: ignore[arg-type] + + +@eval_case(name="kit_marker_mentions_ready") +def kit_marker_mentions_ready(result: AgentExecutionResult) -> float: + return 1.0 if "ready" in response_text(result).casefold() else 0.0 + + +@pytest.mark.agent_eval("kit_marker_mentions_ready") +def test_agent_eval_marker_passes_a_known_good_trace(agent_eval: Any) -> None: + report = agent_eval.run(_result("system ready")) + report.assert_passed() + assert report.results[0].score == 1.0 + + +@pytest.mark.agent_eval("kit_marker_mentions_ready") +def test_agent_eval_marker_fails_a_known_bad_trace(agent_eval: Any) -> None: + with pytest.raises(AssertionError, match="kit_marker_mentions_ready"): + agent_eval.run(_result("nope")) + + +def test_agent_eval_requires_a_selection(agent_eval: Any) -> None: + with pytest.raises(RuntimeError, match="No evals"): + agent_eval.run(_result("ready")) + + +def test_agent_eval_explicit_list_can_fail_closed(agent_eval: Any) -> None: + report = agent_eval.run( + _result("refund"), + RequiredTermsEval(["refund"]), + fail=False, + ) + assert report.passed + bad = agent_eval.run(_result("nope"), RequiredTermsEval(["refund"]), fail=False) + assert not bad.passed diff --git a/tests/test_guardrail_eval_reporting.py b/tests/test_guardrail_eval_reporting.py new file mode 100644 index 0000000..7395f05 --- /dev/null +++ b/tests/test_guardrail_eval_reporting.py @@ -0,0 +1,232 @@ +"""Guardrail and eval results render in the JSON and HTML reports.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + AgentTestConfig, + HtmlReportWriter, + JsonReportWriter, + NoProfanityGuardrail, + RequiredTermsEval, + ToolCall, + guardrail, + must_open_pull_request, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace +from agent_test_kit.plugin import pytest_plugin + +pytestmark = pytest.mark.agent_unit + + +def _result() -> AgentExecutionResult: + return AgentExecutionResult( + success=True, + run_id="run_report", + response="What the fuck — I opened the pull request.", + trace=AgentTrace( + tool_calls=[ + ToolCall( + server="github", + name="create_pull_request", + status=ToolCallStatus.SUCCESS, + operation_kind=ToolOperationKind.WRITE, + ) + ] + ), + ) + + +def test_html_report_includes_guardrail_and_eval_sections(tmp_path: Path) -> None: + result = _result() + with pytest.raises(AssertionError): + result.assert_guardrails([NoProfanityGuardrail(), must_open_pull_request(server="github")]) + with pytest.raises(AssertionError): + result.assert_evals(RequiredTermsEval(["refund"], name="mentions_refund")) + + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="report-agent"), + output_path=tmp_path / "report.json", + ) + writer.record_scenario( + name="guardrails and evals", + nodeid="tests/test_report.py::test_guardrails", + passed=False, + status="failed", + duration_ms=5, + markers=["agent_unit"], + execution_result=result, + ) + report = writer.build_report() + scenario = report.scenarios[0] + assert [item.name for item in scenario.guardrails] == [ + "no_profanity", + "must_open_pull_request", + ] + assert scenario.guardrails[0].passed is False + assert scenario.guardrails[0].mode == "offline" + assert scenario.guardrails[1].passed is True + assert scenario.evals[0].name == "mentions_refund" + assert scenario.evals[0].passed is False + assert scenario.evals[0].score == 0.0 + + html = HtmlReportWriter(tmp_path / "report.html").write(report).read_text(encoding="utf-8") + assert "Guardrails (2)" in html + assert "Evals (1)" in html + assert "no_profanity" in html + assert "must_open_pull_request" in html + assert "mentions_refund" in html + assert "Disallowed language" in html + assert "offline" in html + + +def test_report_escapes_guardrail_messages_and_dedupes_checks(tmp_path: Path) -> None: + @guardrail(phase="message", name="markup") + def markup(text: str) -> str: + return "" + + result = AgentExecutionResult(success=True, run_id="run_html", response="hello") + with pytest.raises(AssertionError): + result.assert_guardrails(markup) + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="report-agent"), + output_path=tmp_path / "report.json", + ) + writer.record_scenario( + name="", + nodeid="tests/demo.py::test_demo", + passed=False, + duration_ms=1, + markers=[], + execution_result=result, + guardrail_checks=list(result.guardrail_checks), + ) + scenario = writer.build_report().scenarios[0] + assert len(scenario.guardrails) == 1 + html = ( + HtmlReportWriter(tmp_path / "report.html") + .write(writer.build_report()) + .read_text(encoding="utf-8") + ) + assert "" not in html + assert "<script>alert(1)</script>" in html + + +def test_plugin_records_guardrails_and_evals_from_fixture_state(tmp_path: Path) -> None: + from unittest.mock import MagicMock + + result = AgentExecutionResult(success=True, run_id="run_plugin", response="hello refund") + evaluation = result.assert_evals(RequiredTermsEval(["refund"])) + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="demo"), + output_path=tmp_path / "report.json", + ) + state = pytest_plugin.PluginState( + agent_config=AgentTestConfig(), + json_writer=writer, + results={"tests/demo.py::test_demo": result}, + eval_checks={"tests/demo.py::test_demo": list(evaluation.results)}, + ) + item = MagicMock() + item.nodeid = "tests/demo.py::test_demo" + item.name = "test_demo" + item.iter_markers.return_value = [] + reports: dict[str, MagicMock] = {} + for phase in ("setup", "call", "teardown"): + report = MagicMock() + report.when = phase + report.duration = 0.01 + report.skipped = False + report.failed = False + report.passed = True + report.longrepr = None + reports[phase] = report + pytest_plugin._record_scenario_from_reports(state, item, reports) + scenario = writer.scenarios[0] + assert len(scenario.evals) == 1 + assert scenario.evals[0].passed is True + assert scenario.evals[0].score == 1.0 + payload = json.loads(writer.write().read_text(encoding="utf-8")) + assert payload["scenarios"][0]["evals"][0]["name"] == "required_terms" + + +def test_html_omits_guardrail_and_eval_sections_when_absent(tmp_path: Path) -> None: + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="empty"), + output_path=tmp_path / "report.json", + ) + writer.record_scenario( + name="plain", + nodeid="tests/plain.py::test_plain", + passed=True, + duration_ms=1, + markers=[], + ) + html = ( + HtmlReportWriter(tmp_path / "report.html") + .write(writer.build_report()) + .read_text(encoding="utf-8") + ) + assert "Guardrails (" not in html + assert "Evals (" not in html + + +def test_pytest_session_html_includes_guardrails_and_evals( + pytester: pytest.Pytester, + tmp_path: Path, +) -> None: + html_path = tmp_path / "report.html" + pytester.makepyfile( + """ +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + NoProfanityGuardrail, + RequiredTermsEval, + ToolCall, + must_open_pull_request, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace + + +@pytest.mark.agent_unit +def test_recorded_run(agent_guardrails, agent_eval): + result = AgentExecutionResult( + success=True, + run_id="run_1", + response="I opened the pull request.", + trace=AgentTrace( + tool_calls=[ + ToolCall( + server="github", + name="create_pull_request", + status=ToolCallStatus.SUCCESS, + operation_kind=ToolOperationKind.WRITE, + ) + ] + ), + ) + agent_guardrails.check( + result, + [NoProfanityGuardrail(), must_open_pull_request(server="github")], + ) + agent_eval.run(result, [RequiredTermsEval(["pull request"])]) +""" + ) + result = pytester.runpytest(f"--agent-report-html={html_path}", "-q") + result.assert_outcomes(passed=1) + html = html_path.read_text(encoding="utf-8") + assert "Guardrails (2)" in html + assert "no_profanity" in html + assert "must_open_pull_request" in html + assert "Evals (1)" in html + assert "required_terms" in html + assert "pull request" in html or "1.00" in html diff --git a/tests/test_guardrails.py b/tests/test_guardrails.py new file mode 100644 index 0000000..de4f149 --- /dev/null +++ b/tests/test_guardrails.py @@ -0,0 +1,333 @@ +"""Guardrails: offline traces, live hooks, and known-good / known-bad traces.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + Guardrail, + GuardrailEvent, + GuardrailMonitor, + GuardrailViolationError, + NoProfanityGuardrail, + RequiredFinalToolGuardrail, + ToolCall, + check_guardrails, + guardrail, + must_open_pull_request, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.trace import AgentTrace +from agent_test_kit.plugin import pytest_plugin +from agent_test_kit.scoring import response_text + +pytestmark = pytest.mark.agent_unit + + +def _result( + response: str = "", + calls: list[ToolCall] | None = None, + *, + success: bool = True, +) -> AgentExecutionResult: + return AgentExecutionResult( + success=success, + run_id="run-guard", + response=response, + trace=AgentTrace(tool_calls=list(calls or [])), + ) + + +def _call( + name: str, + *, + server: str | None = None, + status: ToolCallStatus = ToolCallStatus.SUCCESS, +) -> ToolCall: + return ToolCall( + server=server, + name=name, + status=status, + operation_kind=ToolOperationKind.WRITE, + ) + + +def _run_both(guard: Guardrail, result: AgentExecutionResult) -> tuple[bool, bool]: + """One guardrail definition, offline against the trace and online as a stream.""" + offline = check_guardrails(result, guard) + monitor = GuardrailMonitor([guard]) + text = response_text(result) + if text: + monitor.on_message(text) + for call in result.tool_calls: + monitor.on_tool_call(call) + online = monitor.finish(result) + return offline.passed, online.passed + + +@pytest.mark.parametrize( + ("response", "expect_pass"), + [ + ("I opened the pull request.", True), + ("What the fuck is this diff?", False), + ("SHIT, that failed.", False), + ], +) +def test_no_profanity_guardrail_known_traces(response: str, expect_pass: bool) -> None: + guard = NoProfanityGuardrail() + offline_passed, online_passed = _run_both(guard, _result(response)) + assert offline_passed is expect_pass + assert online_passed is expect_pass + report = check_guardrails(_result(response), guard) + if expect_pass: + report.assert_passed() + else: + with pytest.raises(AssertionError, match="no_profanity"): + report.assert_passed() + + +@pytest.mark.parametrize( + ("calls", "expect_pass"), + [ + ( + [_call("get_diff", server="github"), _call("create_pull_request", server="github")], + True, + ), + ( + [_call("create_pull_request", server="github"), _call("add_comment", server="github")], + False, + ), + ([], False), + ( + [_call("create_pull_request", server="github", status=ToolCallStatus.ERROR)], + False, + ), + ], +) +def test_required_final_tool_guardrail_known_traces( + calls: list[ToolCall], + expect_pass: bool, +) -> None: + guard = RequiredFinalToolGuardrail("create_pull_request", server="github") + offline_passed, online_passed = _run_both(guard, _result("done", calls)) + assert offline_passed is expect_pass + assert online_passed is expect_pass + report = check_guardrails(_result("done", calls), guard) + if expect_pass: + report.assert_passed() + else: + with pytest.raises(AssertionError, match="required_final_tool"): + report.assert_passed() + + +@pytest.mark.parametrize( + ("calls", "expect_pass"), + [ + ([_call("get_diff", server="github"), _call("create_pull_request", server="github")], True), + ([_call("create_pull_request", server="github"), _call("add_comment")], False), + ([], False), + ], +) +def test_must_open_pull_request_known_traces(calls: list[ToolCall], expect_pass: bool) -> None: + guard = must_open_pull_request(server="github") + offline_passed, online_passed = _run_both(guard, _result("I opened the pull request.", calls)) + assert offline_passed is expect_pass + assert online_passed is expect_pass + report = check_guardrails(_result("opening", calls), [guard]) + if expect_pass: + report.assert_passed() + else: + with pytest.raises(AssertionError, match="must_open_pull_request"): + report.assert_passed() + + +def test_online_profanity_fails_fast_on_the_stream() -> None: + monitor = GuardrailMonitor([NoProfanityGuardrail()], fail_fast=True) + with pytest.raises(GuardrailViolationError, match="no_profanity") as caught: + monitor.handle({"type": "message", "role": "assistant", "text": "fuck this"}) + assert caught.value.violation.phase == "message" + assert monitor.last_report is not None + assert not monitor.last_report.passed + + +def test_online_pull_request_rule_uses_the_same_guardrail() -> None: + monitor = GuardrailMonitor([must_open_pull_request(server="github")]) + monitor.handle({"type": "message", "content": "I opened the pull request."}) + monitor.handle( + { + "type": "tool_call", + "server": "github", + "name": "create_pull_request", + "status": "success", + "operation_kind": "write", + } + ) + report = monitor.finish() + report.assert_passed() + assert report.mode == "online" + assert report.checks[0].name == "must_open_pull_request" + + +def test_online_split_profanity_is_caught_when_the_run_finishes() -> None: + monitor = GuardrailMonitor([NoProfanityGuardrail()]) + monitor.on_message("fu") + monitor.on_message("ck") + report = monitor.finish() + assert not report.passed + assert report.checks[0].message is not None + assert "fuck" in report.checks[0].message + + +def test_user_messages_are_not_agent_output() -> None: + monitor = GuardrailMonitor([NoProfanityGuardrail()]) + monitor.on_message("fuck", role="user") + monitor.finish().assert_passed() + + +def test_word_boundaries_and_custom_disallowed_language() -> None: + check_guardrails(_result("classic classification"), NoProfanityGuardrail()).assert_passed() + check_guardrails(_result("a shitake mushroom"), NoProfanityGuardrail()).assert_passed() + custom = NoProfanityGuardrail(terms=["banana"], extra_terms=["system prompt"]) + check_guardrails(_result("fuck"), custom).assert_passed() + bad = check_guardrails(_result("leaked the system prompt"), custom) + assert not bad.passed + assert "system prompt" in (bad.checks[0].message or "") + + +def test_profanity_terms_must_be_non_empty() -> None: + with pytest.raises(ValueError, match="at least one"): + NoProfanityGuardrail(terms=[]) + with pytest.raises(ValueError, match="non-empty"): + NoProfanityGuardrail(extra_terms=[" "]) + + +def test_final_tool_wrong_server_and_optional_success() -> None: + calls = [_call("create_pull_request", server="bitbucket")] + report = check_guardrails( + _result("done", calls), + RequiredFinalToolGuardrail("create_pull_request", server="github"), + ) + assert not report.passed + assert "bitbucket/create_pull_request" in (report.checks[0].message or "") + allowed = RequiredFinalToolGuardrail( + "create_pull_request", + require_success=False, + ) + errored = [_call("create_pull_request", status=ToolCallStatus.ERROR)] + check_guardrails(_result("done", errored), allowed).assert_passed() + + +def test_assert_guardrails_records_a_failure_before_raising() -> None: + result = _result("fuck") + with pytest.raises(AssertionError, match="no_profanity"): + result.assert_guardrails(NoProfanityGuardrail()) + assert result.assertion_outcomes[-1].name == "assert_guardrails" + assert result.assertion_outcomes[-1].passed is False + assert result.guardrail_checks[0].passed is False + + +def test_assert_guardrails_passes_and_records() -> None: + calls = [_call("create_pull_request", server="github")] + result = _result("I opened the pull request.", calls) + report = result.assert_guardrails( + [NoProfanityGuardrail(), must_open_pull_request(server="github")] + ) + assert report.passed + assert result.assertion_outcomes[-1].passed is True + assert [check.name for check in result.guardrail_checks] == [ + "no_profanity", + "must_open_pull_request", + ] + + +def test_custom_guardrail_decorator_and_subclass_share_both_modes() -> None: + @guardrail(phase="message") + def no_secrets(text: str, *, role: str = "assistant") -> str | None: + if role == "assistant" and "sk-" in text: + return "assistant output looks like a secret key" + return None + + class NoDeleteGuardrail(Guardrail): + name = "no_delete" + + def on_tool_call(self, call: ToolCall) -> str | None: + if call.name.startswith("delete_"): + return f"delete tool is not allowed: {call.name}" + return None + + @guardrail + def needs_success(result: AgentExecutionResult) -> str | None: + if not result.success: + return "run failed" + return None + + secret = _result("token sk-live") + assert _run_both(no_secrets, secret) == (False, False) + deleted = _result("ok", [_call("delete_issue")]) + assert _run_both(NoDeleteGuardrail(), deleted) == (False, False) + clean = _result("ok", [_call("create_pull_request")]) + assert _run_both(no_secrets, clean) == (True, True) + assert _run_both(needs_success, _result("ok", success=False)) == (False, False) + no_role = guardrail(phase="message", name="no_email")( + lambda text: "email" if "@" in text else None + ) + assert not check_guardrails(_result("a@b.test"), no_role).passed + + +def test_guardrail_return_type_and_empty_selection() -> None: + @guardrail(name="bad_return") + def bad_return(result: AgentExecutionResult) -> Any: + return 1 + + with pytest.raises(TypeError, match="bad_return"): + check_guardrails(_result("ok"), bad_return) + with pytest.raises(ValueError, match="at least one"): + check_guardrails(_result("ok"), []) + with pytest.raises(ValueError, match="at least one"): + GuardrailMonitor([]) + with pytest.raises(TypeError, match="Guardrail"): + check_guardrails(_result("ok"), ["nope"]) # type: ignore[arg-type] + + +def test_monitor_rejects_a_second_finish_and_unknown_events() -> None: + monitor = GuardrailMonitor(NoProfanityGuardrail()) + monitor.finish() + with pytest.raises(RuntimeError, match="already"): + monitor.finish() + with pytest.raises(RuntimeError, match="already"): + monitor.on_message("hi") + fresh = GuardrailMonitor(NoProfanityGuardrail()) + with pytest.raises(ValueError, match="Unknown"): + fresh.handle({"type": "pause"}) + with pytest.raises(ValueError, match="name"): + fresh.handle({"type": "tool_call"}) + event = GuardrailEvent( + type="tool_call", + tool_call=_call("create_pull_request", server="github"), + ) + assert fresh.handle(event) == [] + + +def test_fixture_records_online_violation_for_the_report( + agent_guardrails: Any, + request: pytest.FixtureRequest, +) -> None: + monitor = agent_guardrails.monitor([NoProfanityGuardrail()], fail_fast=True) + with pytest.raises(GuardrailViolationError): + monitor.on_message("fuck") + agent_guardrails.close() + state = request.config.stash[pytest_plugin._STASH_KEY] + checks = state.guardrail_checks[request.node.nodeid] + assert checks[0].mode == "online" + assert checks[0].passed is False + assert "fuck" in (checks[0].message or "") + + +def test_fixture_offline_check_fails_a_known_bad_trace(agent_guardrails: Any) -> None: + result = _result("fuck") + with pytest.raises(AssertionError, match="no_profanity"): + agent_guardrails.check(result, NoProfanityGuardrail()) + assert result.guardrail_checks[0].passed is False diff --git a/tests/test_pytest_plugin.py b/tests/test_pytest_plugin.py index 6660d83..1c67c9c 100644 --- a/tests/test_pytest_plugin.py +++ b/tests/test_pytest_plugin.py @@ -148,7 +148,9 @@ def test_agent_report_html_includes_tool_flow_from_fixture( asyncio_mode = auto """ ) - pytester.makeconftest( + # The fake client lives in the test module. The runtime plugin is registered + # at session start, so a conftest fixture of the same name does not win. + pytester.makepyfile( """ import httpx import pytest @@ -186,11 +188,6 @@ def agent_client(agent_test_config, request): result, ), ) -""" - ) - pytester.makepyfile( - """ -import pytest @pytest.mark.asyncio