diff --git a/README.md b/README.md index 116ab2c..b367de3 100644 --- a/README.md +++ b/README.md @@ -308,6 +308,52 @@ aggregate.assert_responses_stable() aggregate.assert_traces_stable() ``` +### Behavior scoring (no single expected result) + +AI output can vary between runs and still be correct. Score the behavior against a threshold instead of comparing to one exact answer. + +| API | What it checks | +|-----|----------------| +| `await result.assert_score(scorer, *, min_score)` | One run scores ≥ `min_score` (0.0–1.0); score and reason appear in the report | +| `aggregate.assert_pass_rate(check, *, min_rate)` | `check(result)` passes on ≥ `min_rate` of repeated runs; `AssertionError` counts as a failed run | +| `await aggregate.assert_score_rate(scorer, *, min_score, min_rate)` | Scorer reaches `min_score` on ≥ `min_rate` of repeated runs | + +Built-in scorers are deterministic and dependency-free: + +| Scorer | Score | +|--------|-------| +| `RequiredTermsScorer(terms)` | Fraction of terms present in the response | +| `ForbiddenTermsScorer(terms)` | 1.0 if none present, else 0.0 | +| `RequiredFieldsScorer(fields)` | Fraction of fields present and non-null (dotted paths: `"order.status"`) | + +```python +from agent_test_kit import ForbiddenTermsScorer, RequiredTermsScorer, run_repeatedly + +await result.assert_score(RequiredTermsScorer(["order 123", "pending"]), min_score=1.0) +await result.assert_score(ForbiddenTermsScorer(["system prompt", "api key"]), min_score=1.0) + +aggregate = await run_repeatedly(agent_client, input={"order_id": 123}, runs=5, idempotency_key="k") +aggregate.assert_pass_rate(lambda r: r.success, min_rate=0.8) +await aggregate.assert_score_rate(RequiredTermsScorer(["pending"]), min_score=1.0, min_rate=0.8) +``` + +Judgment-based scoring (LLM judge, embeddings, rubric) lives in **your** agent repo. Implement the `Scorer` protocol; `score` may be sync or async: + +```python +from agent_test_kit import Score, response_text + + +class GroundednessJudge: + name = "groundedness" + + async def score(self, result) -> Score: + verdict = await my_llm_judge(response_text(result)) + return Score(verdict.value, reason=verdict.rationale) + + +await result.assert_score(GroundednessJudge(), min_score=0.8) +``` + ### Regression cases (YAML-driven) Load cases with `load_regression_cases(path)`; evaluate with `case.evaluate(result)`. @@ -621,14 +667,20 @@ from agent_test_kit import ( AgentExecutionResult, AnyOfStep, CleanupManager, + ForbiddenTermsScorer, HtmlReportWriter, JsonReportWriter, OptionalStep, PromptReviewClient, RepeatedExecutionResult, + RequiredFieldsScorer, + RequiredTermsScorer, + Score, + Scorer, ToolStep, get_profile, load_regression_cases, + response_text, run_repeatedly, run_verifiers, SideEffectVerifier, diff --git a/examples/consumer_test_example.py b/examples/consumer_test_example.py index 3c46786..43e0d1c 100644 --- a/examples/consumer_test_example.py +++ b/examples/consumer_test_example.py @@ -4,6 +4,8 @@ import pytest +from agent_test_kit import ForbiddenTermsScorer, RequiredFieldsScorer, run_repeatedly + @pytest.mark.asyncio @pytest.mark.agent_integration @@ -21,3 +23,29 @@ async def test_agent_flow(agent_client: object) -> None: result.assert_tool_called("bitbucket_get_diff", server="bitbucket") result.assert_tool_called("jira_create_issue", server="jira") result.assert_tool_not_called("bitbucket_merge_pull_request", server="bitbucket") + + +@pytest.mark.asyncio +@pytest.mark.agent_integration +@pytest.mark.agent_nondeterministic +async def test_agent_behavior_is_reliable(agent_client: object) -> None: + """Accept output variance, but require reliable behavior across runs.""" + aggregate = await run_repeatedly( + agent_client, # type: ignore[arg-type] + input={"repository": "test-repository", "pull_request_id": 125}, + runs=5, + idempotency_key="pr-125-review", + ) + + aggregate.assert_no_duplicate_writes() + aggregate.assert_pass_rate(lambda result: result.success, min_rate=0.8) + await aggregate.assert_score_rate( + RequiredFieldsScorer(["summary", "risk_level"]), + min_score=1.0, + min_rate=0.8, + ) + await aggregate.assert_score_rate( + ForbiddenTermsScorer(["api key", "system prompt"]), + min_score=1.0, + min_rate=1.0, + ) diff --git a/src/agent_test_kit/__init__.py b/src/agent_test_kit/__init__.py index 54b5297..6e27a3f 100644 --- a/src/agent_test_kit/__init__.py +++ b/src/agent_test_kit/__init__.py @@ -38,6 +38,14 @@ from agent_test_kit.reporting.html_report import HtmlReportWriter from agent_test_kit.reporting.json_report import JsonReportWriter from agent_test_kit.reporting.markdown_status import MarkdownStatusReportWriter +from agent_test_kit.scoring import ( + ForbiddenTermsScorer, + RequiredFieldsScorer, + RequiredTermsScorer, + Score, + Scorer, + response_text, +) from agent_test_kit.verifiers import ( SideEffectVerifier, VerificationContext, @@ -57,6 +65,7 @@ "CleanupManager", "CursorAgentClient", "FlowAssertionEngine", + "ForbiddenTermsScorer", "HtmlReportWriter", "HttpResultPublisher", "JsonReportWriter", @@ -71,10 +80,14 @@ "RegressionExpectedOutcome", "RegressionToolExpectation", "RepeatedExecutionResult", + "RequiredFieldsScorer", + "RequiredTermsScorer", "ResultPublishError", "ResultPublisher", "ResultPublisherConfig", "RunContext", + "Score", + "Scorer", "SideEffectVerifier", "ToolStep", "VerificationContext", @@ -83,6 +96,7 @@ "__version__", "load_regression_cases", "get_profile", + "response_text", "run_repeatedly", "run_verifiers", ] diff --git a/src/agent_test_kit/models/execution.py b/src/agent_test_kit/models/execution.py index a47a0d1..875d0c1 100644 --- a/src/agent_test_kit/models/execution.py +++ b/src/agent_test_kit/models/execution.py @@ -19,6 +19,7 @@ WorkflowStep, ) from agent_test_kit.models.enums import ToolCallStatus + from agent_test_kit.scoring import Score, Scorer class AssertionOutcome(BaseModel): @@ -335,6 +336,29 @@ def assert_write_tool_called_once( ), ) + async def assert_score(self, scorer: Scorer, *, min_score: float) -> Score: + """Score this result and require ``score.value >= min_score``.""" + from agent_test_kit.scoring import evaluate_score, validate_unit_interval + + validate_unit_interval("min_score", min_score) + score = await evaluate_score(scorer, self) + + def evaluate() -> None: + if score.value < min_score: + detail = f" ({score.reason})" if score.reason else "" + raise AssertionError( + f"Expected {scorer.name} score >= {min_score:.2f}. " + f"Observed: {score.value:.2f}{detail}" + ) + + self._capture_assertion( + "assert_score", + expected={"scorer": scorer.name, "min_score": min_score}, + actual={"score": score.value, "reason": score.reason, "details": score.details}, + assertion=evaluate, + ) + return score + @classmethod def from_api_response( cls, diff --git a/src/agent_test_kit/repetition.py b/src/agent_test_kit/repetition.py index 76508fa..e3b3907 100644 --- a/src/agent_test_kit/repetition.py +++ b/src/agent_test_kit/repetition.py @@ -11,8 +11,10 @@ from agent_test_kit.models.run_context import RunContext from agent_test_kit.models.tool_call import ToolCall from agent_test_kit.models.trace import AgentTrace +from agent_test_kit.scoring import Scorer, evaluate_score, validate_unit_interval ContextFactory = Callable[[int, str], RunContext] +RunCheck = Callable[[AgentExecutionResult], bool] AsyncExecutionCallable = Callable[ [dict[str, Any], RunContext], Awaitable[AgentExecutionResult], @@ -105,6 +107,45 @@ def assert_traces_stable( label="traces", ) + def assert_pass_rate(self, check: RunCheck, *, min_rate: float) -> float: + """Require ``check`` to pass on at least ``min_rate`` of runs; return the rate. + + A check passes by returning True. Returning False or raising AssertionError + counts as a failed run, so existing ``result.assert_*`` calls can be reused. + """ + validate_unit_interval("min_rate", min_rate) + outcomes = [_run_check(check, result) for result in self.results] + return self._assert_rate(outcomes, min_rate, label="check") + + async def assert_score_rate( + self, + scorer: Scorer, + *, + min_score: float, + min_rate: float, + ) -> float: + """Require ``scorer`` to reach ``min_score`` on at least ``min_rate`` of runs.""" + validate_unit_interval("min_score", min_score) + validate_unit_interval("min_rate", min_rate) + outcomes = [ + (await evaluate_score(scorer, result)).value >= min_score for result in self.results + ] + return self._assert_rate(outcomes, min_rate, label=f"{scorer.name} >= {min_score:.2f}") + + @staticmethod + def _assert_rate(outcomes: list[bool], min_rate: float, *, label: str) -> float: + if not outcomes: + raise ValueError("pass rate requires at least one run") + passed = sum(outcomes) + rate = passed / len(outcomes) + if rate < min_rate: + failing_runs = [index for index, ok in enumerate(outcomes, start=1) if not ok] + raise AssertionError( + f"Expected {label} pass rate >= {min_rate:.0%}. " + f"Observed: {rate:.0%} ({passed}/{len(outcomes)}); failing runs: {failing_runs}" + ) + return rate + @staticmethod def _assert_stable_values( values: list[Any], @@ -158,6 +199,13 @@ async def run_repeatedly( ) +def _run_check(check: RunCheck, result: AgentExecutionResult) -> bool: + try: + return bool(check(result)) + except AssertionError: + return False + + def _default_context_factory(_index: int, idempotency_key: str) -> RunContext: return RunContext(idempotency_key=idempotency_key) diff --git a/src/agent_test_kit/scoring.py b/src/agent_test_kit/scoring.py new file mode 100644 index 0000000..6de926e --- /dev/null +++ b/src/agent_test_kit/scoring.py @@ -0,0 +1,168 @@ +"""Behavior scoring for agent outputs that have no single expected result.""" + +from __future__ import annotations + +import inspect +import json +import math +from collections.abc import Awaitable, Iterable, Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +if TYPE_CHECKING: + from agent_test_kit.models.execution import AgentExecutionResult + + +@dataclass(frozen=True, slots=True) +class Score: + """Normalized score in the closed interval [0.0, 1.0].""" + + value: float + reason: str | None = None + details: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + validate_unit_interval("score value", self.value) + + +@runtime_checkable +class Scorer(Protocol): + """Scores one execution result; sync or async implementations are both accepted. + + Consumer repositories implement judgment-based scorers (LLM judge, embeddings, + rubric review). The package ships only deterministic scorers. + """ + + name: str + + def score(self, result: AgentExecutionResult) -> Score | Awaitable[Score]: + """Return a score for ``result``.""" + ... + + +async def evaluate_score(scorer: Scorer, result: AgentExecutionResult) -> Score: + """Run a sync or async scorer and validate its return type.""" + outcome = scorer.score(result) + if inspect.isawaitable(outcome): + outcome = await outcome + if not isinstance(outcome, Score): + raise TypeError( + f"Scorer {scorer.name!r} must return Score. Observed: {type(outcome).__name__}" + ) + return outcome + + +def validate_unit_interval(label: str, value: float) -> None: + """Reject values outside [0.0, 1.0], including NaN and booleans.""" + if isinstance(value, bool) or not isinstance(value, int | float): + raise TypeError(f"{label} must be a number between 0.0 and 1.0") + if math.isnan(value) or not 0.0 <= value <= 1.0: + raise ValueError(f"{label} must be between 0.0 and 1.0. Observed: {value!r}") + + +def response_text(result: AgentExecutionResult) -> str: + """Render ``result.response`` as text for term-based scoring.""" + response = result.response + if response is None: + return "" + if isinstance(response, str): + return response + return json.dumps(response, sort_keys=True, default=str, ensure_ascii=False) + + +class RequiredTermsScorer: + """Fraction of required terms present in the response text.""" + + def __init__( + self, + terms: Iterable[str], + *, + case_sensitive: bool = False, + name: str = "required_terms", + ) -> None: + self.terms = _validated_terms(terms) + self.case_sensitive = case_sensitive + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + text = _normalize(response_text(result), self.case_sensitive) + missing = [term for term in self.terms if _normalize(term, self.case_sensitive) not in text] + found = len(self.terms) - len(missing) + return Score( + found / len(self.terms), + reason=f"Missing terms: {missing}" if missing else None, + details={"missing": missing}, + ) + + +class ForbiddenTermsScorer: + """Scores 1.0 when no forbidden term appears in the response, otherwise 0.0.""" + + def __init__( + self, + terms: Iterable[str], + *, + case_sensitive: bool = False, + name: str = "forbidden_terms", + ) -> None: + self.terms = _validated_terms(terms) + self.case_sensitive = case_sensitive + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + text = _normalize(response_text(result), self.case_sensitive) + found = [term for term in self.terms if _normalize(term, self.case_sensitive) in text] + return Score( + 0.0 if found else 1.0, + reason=f"Forbidden terms present: {found}" if found else None, + details={"found": found}, + ) + + +class RequiredFieldsScorer: + """Fraction of required fields present and non-null in a mapping response. + + Fields support dotted paths such as ``"order.status"``. + """ + + def __init__(self, fields: Iterable[str], *, name: str = "required_fields") -> None: + self.fields = _validated_terms(fields) + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + response = result.response + if not isinstance(response, Mapping): + return Score( + 0.0, + reason="Response is not a JSON object", + details={"missing": list(self.fields)}, + ) + missing = [path for path in self.fields if _lookup(response, path) is None] + present = len(self.fields) - len(missing) + return Score( + present / len(self.fields), + reason=f"Missing fields: {missing}" if missing else None, + details={"missing": missing}, + ) + + +def _validated_terms(terms: Iterable[str]) -> tuple[str, ...]: + values = tuple(terms) + if not values: + raise ValueError("at least one term is required") + if any(not isinstance(value, str) or not value for value in values): + raise ValueError("terms must be non-empty strings") + return values + + +def _normalize(text: str, case_sensitive: bool) -> str: + return text if case_sensitive else text.casefold() + + +def _lookup(payload: Mapping[str, Any], path: str) -> Any: + current: Any = payload + for part in path.split("."): + if not isinstance(current, Mapping) or part not in current: + return None + current = current[part] + return current diff --git a/tests/test_scoring.py b/tests/test_scoring.py new file mode 100644 index 0000000..a522d41 --- /dev/null +++ b/tests/test_scoring.py @@ -0,0 +1,372 @@ +"""Behavior scoring contract tests.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + AgentTestConfig, + ForbiddenTermsScorer, + JsonReportWriter, + RepeatedExecutionResult, + RequiredFieldsScorer, + RequiredTermsScorer, + RunContext, + Score, + Scorer, + response_text, + run_repeatedly, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.tool_call import ToolCall +from agent_test_kit.models.trace import AgentTrace + +pytestmark = pytest.mark.agent_unit + + +def _result(response: Any, *, success: bool = True) -> AgentExecutionResult: + return AgentExecutionResult(success=success, run_id="run-1", response=response) + + +class FixedScorer: + """Async scorer that returns a preset value, standing in for an LLM judge.""" + + def __init__(self, value: float, *, name: str = "judge") -> None: + self.value = value + self.name = name + + async def score(self, result: AgentExecutionResult) -> Score: + return Score(self.value, reason="judge rationale") + + +class SequenceScorer: + """Sync scorer returning successive values across repeated runs.""" + + name = "sequence" + + def __init__(self, values: list[float]) -> None: + self._values = iter(values) + + def score(self, result: AgentExecutionResult) -> Score: + return Score(next(self._values)) + + +class BadScorer: + name = "bad" + + def score(self, result: AgentExecutionResult) -> Any: + return 0.9 + + +class ScriptedExecutor: + """Returns one scripted response per run.""" + + def __init__(self, responses: list[Any], *, write_on_run: int | None = None) -> None: + self._responses = iter(responses) + self._run = 0 + self._write_on_run = write_on_run + + async def execute( + self, + input: dict[str, Any], + *, + run_context: RunContext | None = None, + ) -> AgentExecutionResult: + self._run += 1 + calls = [] + if self._run == self._write_on_run: + calls.append( + ToolCall( + server="billing", + name="issue_refund", + operation_kind=ToolOperationKind.WRITE, + status=ToolCallStatus.SUCCESS, + ) + ) + return AgentExecutionResult( + success=True, + run_id=f"run-{self._run}", + response=next(self._responses), + trace=AgentTrace(tool_calls=calls), + ) + + +@pytest.mark.parametrize( + ("response", "expected"), + [ + (None, ""), + ("plain text", "plain text"), + ({"b": 1, "a": "שלום"}, '{"a": "שלום", "b": 1}'), + (["x", 2], '["x", 2]'), + ], +) +def test_response_text_renders_supported_shapes(response: Any, expected: str) -> None: + assert response_text(_result(response)) == expected + + +@pytest.mark.parametrize("value", [-0.01, 1.01, float("nan")]) +def test_score_rejects_values_outside_unit_interval(value: float) -> None: + with pytest.raises(ValueError, match="between 0.0 and 1.0"): + Score(value) + + +@pytest.mark.parametrize("value", [True, "0.5", None]) +def test_score_rejects_non_numeric_values(value: Any) -> None: + with pytest.raises(TypeError, match="must be a number"): + Score(value) + + +@pytest.mark.parametrize( + ("response", "expected_value", "missing"), + [ + ("Refund for ORDER 123 is Pending", 1.0, []), + ("Order 123 received", 0.5, ["pending"]), + ({"note": "nothing useful"}, 0.0, ["order 123", "pending"]), + ], +) +def test_required_terms_scores_fraction_found( + response: Any, expected_value: float, missing: list[str] +) -> None: + score = RequiredTermsScorer(["order 123", "pending"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"missing": missing} + assert (score.reason is None) == (not missing) + + +def test_required_terms_can_be_case_sensitive() -> None: + scorer = RequiredTermsScorer(["Pending"], case_sensitive=True) + + assert scorer.score(_result("status: pending")).value == 0.0 + assert scorer.score(_result("status: Pending")).value == 1.0 + + +@pytest.mark.parametrize( + ("response", "expected_value", "found"), + [ + ("Your refund is being reviewed", 1.0, []), + ("Here is the API KEY: abc", 0.0, ["api key"]), + ({"debug": "system prompt and api key"}, 0.0, ["system prompt", "api key"]), + ], +) +def test_forbidden_terms_is_binary(response: Any, expected_value: float, found: list[str]) -> None: + score = ForbiddenTermsScorer(["system prompt", "api key"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"found": found} + + +@pytest.mark.parametrize( + ("response", "expected_value", "missing"), + [ + ({"order": {"id": 123, "status": "pending"}}, 1.0, []), + ({"order": {"id": 123, "status": None}}, 0.5, ["order.status"]), + ({"order": "123"}, 0.0, ["order.id", "order.status"]), + ], +) +def test_required_fields_supports_dotted_paths( + response: Any, expected_value: float, missing: list[str] +) -> None: + score = RequiredFieldsScorer(["order.id", "order.status"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"missing": missing} + + +def test_required_fields_rejects_non_mapping_response() -> None: + score = RequiredFieldsScorer(["status"]).score(_result("status: ok")) + + assert score.value == 0.0 + assert score.reason == "Response is not a JSON object" + + +@pytest.mark.parametrize( + "factory", + [RequiredTermsScorer, ForbiddenTermsScorer, RequiredFieldsScorer], +) +@pytest.mark.parametrize("terms", [[], [""]]) +def test_builtin_scorers_reject_empty_terms(factory: Any, terms: list[str]) -> None: + with pytest.raises(ValueError): + factory(terms) + + +def test_builtin_and_custom_scorers_satisfy_protocol() -> None: + assert isinstance(RequiredTermsScorer(["x"]), Scorer) + assert isinstance(FixedScorer(0.5), Scorer) + + +@pytest.mark.asyncio +async def test_assert_score_records_passing_outcome() -> None: + result = _result("Order 123 is pending") + + score = await result.assert_score(RequiredTermsScorer(["order 123"]), min_score=1.0) + + assert score.value == 1.0 + outcome = result.assertion_outcomes[-1] + assert outcome.name == "assert_score" + assert outcome.passed is True + assert outcome.expected == {"scorer": "required_terms", "min_score": 1.0} + assert outcome.actual == {"score": 1.0, "reason": None, "details": {"missing": []}} + + +@pytest.mark.asyncio +async def test_assert_score_fails_below_threshold_with_reason() -> None: + result = _result("anything") + + with pytest.raises(AssertionError, match=r"judge score >= 0\.80\. Observed: 0\.60"): + await result.assert_score(FixedScorer(0.6), min_score=0.8) + + outcome = result.assertion_outcomes[-1] + assert outcome.passed is False + assert outcome.actual["score"] == 0.6 + assert "judge rationale" in (outcome.message or "") + + +@pytest.mark.asyncio +async def test_assert_score_passes_at_exact_threshold() -> None: + await _result("x").assert_score(FixedScorer(0.8), min_score=0.8) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("min_score", [-0.1, 1.5]) +async def test_assert_score_rejects_invalid_threshold(min_score: float) -> None: + result = _result("x") + + with pytest.raises(ValueError, match="min_score"): + await result.assert_score(FixedScorer(0.9), min_score=min_score) + assert result.assertion_outcomes == [] + + +@pytest.mark.asyncio +async def test_assert_score_rejects_scorer_with_wrong_return_type() -> None: + with pytest.raises(TypeError, match="must return Score"): + await _result("x").assert_score(BadScorer(), min_score=0.5) + + +@pytest.mark.asyncio +async def test_score_outcome_reaches_json_report(tmp_path: Path) -> None: + result = _result("Order 123") + with pytest.raises(AssertionError): + await result.assert_score(RequiredTermsScorer(["order 123", "pending"]), min_score=0.9) + + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="billing-agent", environment="unit"), + output_path=tmp_path / "report.json", + ) + writer.record_scenario( + name="refund_status", + nodeid="tests/test_x.py::refund_status", + passed=False, + duration_ms=1, + markers=["agent_unit"], + execution_result=result, + ) + report = writer.build_report() + + assertion = report.scenarios[0].assertions[0] + assert assertion.name == "assert_score" + assert assertion.passed is False + assert assertion.actual["score"] == 0.5 + + +@pytest.mark.asyncio +async def test_assert_pass_rate_tolerates_variance_within_threshold() -> None: + responses = ["Order 123 pending", "Order 123 is pending", "cannot find order", "pending: 123"] + aggregate = await run_repeatedly( + ScriptedExecutor(responses), input={"order": 123}, runs=4, idempotency_key="k" + ) + + rate = aggregate.assert_pass_rate( + lambda r: "pending" in response_text(r).casefold(), min_rate=0.75 + ) + + assert rate == 0.75 + + +@pytest.mark.asyncio +async def test_assert_pass_rate_reports_failing_runs() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["ok", "bad", "ok", "bad"]), input={}, runs=4, idempotency_key="k" + ) + + with pytest.raises(AssertionError, match=r"Observed: 50% \(2/4\); failing runs: \[2, 4\]"): + aggregate.assert_pass_rate(lambda r: r.response == "ok", min_rate=0.8) + + +@pytest.mark.asyncio +async def test_assert_pass_rate_counts_assertion_errors_as_failed_runs() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b", "c"], write_on_run=2), input={}, runs=3, idempotency_key="k" + ) + + def read_only(result: AgentExecutionResult) -> bool: + result.assert_read_only() + return True + + assert aggregate.assert_pass_rate(read_only, min_rate=0.6) == pytest.approx(2 / 3) + with pytest.raises(AssertionError, match=r"failing runs: \[2\]"): + aggregate.assert_pass_rate(read_only, min_rate=1.0) + + +@pytest.mark.asyncio +async def test_assert_pass_rate_propagates_unexpected_errors() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor([None]), input={}, runs=1, idempotency_key="k" + ) + + def broken(result: AgentExecutionResult) -> bool: + raise KeyError("missing") + + with pytest.raises(KeyError): + aggregate.assert_pass_rate(broken, min_rate=0.5) + + +@pytest.mark.asyncio +async def test_assert_score_rate_uses_score_threshold_per_run() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b", "c", "d", "e"]), input={}, runs=5, idempotency_key="k" + ) + + rate = await aggregate.assert_score_rate( + SequenceScorer([0.9, 0.85, 0.4, 0.8, 0.95]), min_score=0.8, min_rate=0.8 + ) + + assert rate == 0.8 + + +@pytest.mark.asyncio +async def test_assert_score_rate_fails_below_rate() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b"]), input={}, runs=2, idempotency_key="k" + ) + + with pytest.raises(AssertionError, match=r"sequence >= 0\.80 pass rate >= 100%"): + await aggregate.assert_score_rate(SequenceScorer([0.9, 0.1]), min_score=0.8, min_rate=1.0) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("kwargs", "label"), + [ + ({"min_score": 1.2, "min_rate": 0.5}, "min_score"), + ({"min_score": 0.5, "min_rate": -1.0}, "min_rate"), + ], +) +async def test_assert_score_rate_validates_thresholds(kwargs: dict[str, float], label: str) -> None: + aggregate = await run_repeatedly(ScriptedExecutor(["a"]), input={}, runs=1, idempotency_key="k") + + with pytest.raises(ValueError, match=label): + await aggregate.assert_score_rate(FixedScorer(0.9), **kwargs) + + +def test_pass_rate_requires_results() -> None: + empty = RepeatedExecutionResult(results=(), contexts=(), idempotency_key="k") + + with pytest.raises(ValueError, match="at least one run"): + empty.assert_pass_rate(lambda r: True, min_rate=0.5) + with pytest.raises(ValueError, match="min_rate"): + empty.assert_pass_rate(lambda r: True, min_rate=2.0)