Add behavior scorers and pass-rate assertions

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,

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,
+    )

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",
 ]

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,

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)