From faec7b2f2d73923838c534d0d7c092d45f06085d Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 17:35:31 +0000
Subject: [PATCH 1/6] Add behavior scorers with thresholds and pass-rate
 assertions

Adds a Scorer protocol (sync or async), deterministic built-in scorers
(required terms, forbidden terms, required fields), assert_score on a single
execution, and assert_pass_rate / assert_score_rate on repeated runs. Score
outcomes flow into the existing JSON and HTML report assertion tables.

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 src/agent_test_kit/__init__.py         |  14 +
 src/agent_test_kit/models/execution.py |  24 ++
 src/agent_test_kit/repetition.py       |  48 ++++
 src/agent_test_kit/scoring.py          | 168 +++++++++++
 tests/test_scoring.py                  | 372 +++++++++++++++++++++++++
 5 files changed, 626 insertions(+)
 create mode 100644 src/agent_test_kit/scoring.py
 create mode 100644 tests/test_scoring.py

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)

From 13906074ba28220fa7e9f97f934cababfaf1b53a Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 17:35:31 +0000
Subject: [PATCH 2/6] Document behavior scoring and add repeated-run example

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 README.md                         | 52 +++++++++++++++++++++++++++++++
 examples/consumer_test_example.py | 28 +++++++++++++++++
 2 files changed, 80 insertions(+)

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

From 99adec77c4b3124fa8660ef34a87dd1858fb6628 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 17:42:32 +0000
Subject: [PATCH 3/6] Add goal-based regression cases evaluated across repeated
 runs

Regression cases accept optional goal, persona, runs and behavior fields.
Built-in behavior checks (required/forbidden terms, required fields) apply in
evaluate(); named scores and min_pass_rate apply in evaluate_runs() and run(),
which executes a case repeatedly and reports per-run failures.

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 examples/behavior_regression_cases.yaml |  48 ++++
 src/agent_test_kit/__init__.py          |   4 +
 src/agent_test_kit/regression.py        | 189 ++++++++++++++-
 tests/test_regression_behavior.py       | 298 ++++++++++++++++++++++++
 4 files changed, 530 insertions(+), 9 deletions(-)
 create mode 100644 examples/behavior_regression_cases.yaml
 create mode 100644 tests/test_regression_behavior.py

diff --git a/examples/behavior_regression_cases.yaml b/examples/behavior_regression_cases.yaml
new file mode 100644
index 0000000..822561d
--- /dev/null
+++ b/examples/behavior_regression_cases.yaml
@@ -0,0 +1,48 @@
+# Goal-based regression cases: describe what the user must achieve and what must
+# never happen, then let behavior vary across repeated runs within a pass rate.
+#
+# Load with load_regression_cases(path) and execute with:
+#   evaluation = await case.run(agent_client, scorers=[GroundednessJudge()])
+#   evaluation.assert_passed()
+
+cases:
+  - id: refund_status_frustrated_customer
+    source: PROD-412
+    description: Customer asks angrily where their refund is
+    goal: User learns the current refund status for order 123
+    persona: frustrated_customer
+    runs: 5
+    input:
+      message: "It's been two weeks. Where is my refund for order 123?"
+    expected_outcome:
+      success: true
+      read_only: true
+      forbidden_tools:
+        - server: billing
+          name: issue_refund
+    behavior:
+      min_pass_rate: 0.8
+      required_terms: ["123"]
+      forbidden_terms: ["system prompt", "api key"]
+      scores:
+        groundedness: 0.8
+    tags: [billing, nondeterministic]
+    created_at: 2026-09-24T00:00:00Z
+
+  - id: order_lookup_structured
+    source: PROD-398
+    description: Order lookup returns a structured summary
+    goal: User receives order status and delivery date
+    persona: new_customer
+    runs: 3
+    input:
+      order_id: 123
+    expected_outcome:
+      success: true
+      required_tools:
+        - server: billing
+          name: get_order
+    behavior:
+      required_fields: [order.status, order.delivery_date]
+    tags: [billing]
+    created_at: 2026-09-24T00:00:00Z
diff --git a/src/agent_test_kit/__init__.py b/src/agent_test_kit/__init__.py
index 6e27a3f..b7591b8 100644
--- a/src/agent_test_kit/__init__.py
+++ b/src/agent_test_kit/__init__.py
@@ -23,10 +23,12 @@
     ResultPublishError,
 )
 from agent_test_kit.regression import (
+    RegressionBehavior,
     RegressionCase,
     RegressionCaseLoadError,
     RegressionEvaluation,
     RegressionExpectedOutcome,
+    RegressionRunEvaluation,
     RegressionToolExpectation,
     load_regression_cases,
 )
@@ -74,10 +76,12 @@
     "PromptReviewClient",
     "PromptAiHelperProfile",
     "PublishResult",
+    "RegressionBehavior",
     "RegressionCase",
     "RegressionCaseLoadError",
     "RegressionEvaluation",
     "RegressionExpectedOutcome",
+    "RegressionRunEvaluation",
     "RegressionToolExpectation",
     "RepeatedExecutionResult",
     "RequiredFieldsScorer",
diff --git a/src/agent_test_kit/regression.py b/src/agent_test_kit/regression.py
index 0d5d26c..73f9b59 100644
--- a/src/agent_test_kit/regression.py
+++ b/src/agent_test_kit/regression.py
@@ -3,11 +3,11 @@
 from __future__ import annotations
 
 import json
-from collections.abc import Callable
+from collections.abc import Callable, Iterable, Sequence
 from datetime import datetime
 from functools import partial
 from pathlib import Path
-from typing import Any, Self
+from typing import Annotated, Any, Self
 
 import yaml
 from pydantic import AliasChoices, BaseModel, Field, ValidationError, model_validator
@@ -18,6 +18,21 @@
 )
 from agent_test_kit.models.enums import ToolCallStatus
 from agent_test_kit.models.execution import AgentExecutionResult
+from agent_test_kit.repetition import (
+    AsyncAgentExecutor,
+    AsyncExecutionCallable,
+    run_repeatedly,
+)
+from agent_test_kit.scoring import (
+    ForbiddenTermsScorer,
+    RequiredFieldsScorer,
+    RequiredTermsScorer,
+    Scorer,
+    evaluate_score,
+)
+
+UnitInterval = Annotated[float, Field(ge=0.0, le=1.0)]
+BuiltinScorer = RequiredTermsScorer | ForbiddenTermsScorer | RequiredFieldsScorer
 
 
 class RegressionCaseLoadError(ValueError):
@@ -71,12 +86,44 @@ class RegressionExpectedOutcome(BaseModel):
     model_config = {"extra": "forbid"}
 
 
+class RegressionBehavior(BaseModel):
+    """Behavior expectations for outputs that have no single expected result.
+
+    Built-in checks run in ``evaluate``. Named ``scores`` refer to consumer-provided
+    scorers and are applied by ``evaluate_runs`` and ``run``.
+    """
+
+    min_pass_rate: float = Field(default=1.0, gt=0.0, le=1.0)
+    required_terms: list[str] = Field(default_factory=list)
+    forbidden_terms: list[str] = Field(default_factory=list)
+    required_fields: list[str] = Field(default_factory=list)
+    case_sensitive: bool = False
+    scores: dict[str, UnitInterval] = Field(default_factory=dict)
+
+    model_config = {"extra": "forbid"}
+
+    def builtin_scorers(self) -> list[BuiltinScorer]:
+        scorers: list[BuiltinScorer] = []
+        if self.required_terms:
+            scorers.append(
+                RequiredTermsScorer(self.required_terms, case_sensitive=self.case_sensitive)
+            )
+        if self.forbidden_terms:
+            scorers.append(
+                ForbiddenTermsScorer(self.forbidden_terms, case_sensitive=self.case_sensitive)
+            )
+        if self.required_fields:
+            scorers.append(RequiredFieldsScorer(self.required_fields))
+        return scorers
+
+
 class RegressionEvaluation(BaseModel):
-    """Deterministic evaluation result suitable for ordinary pytest assertions."""
+    """Evaluation of one run, suitable for ordinary pytest assertions."""
 
     case_id: str
     passed: bool
     failures: tuple[str, ...] = ()
+    scores: dict[str, float] = Field(default_factory=dict)
 
     model_config = {"frozen": True}
 
@@ -87,6 +134,35 @@ def assert_passed(self) -> None:
             )
 
 
+class RegressionRunEvaluation(BaseModel):
+    """Evaluation of a case across repeated runs against its minimum pass rate."""
+
+    case_id: str
+    goal: str | None = None
+    persona: str | None = None
+    passed: bool
+    pass_rate: float
+    min_pass_rate: float
+    runs: tuple[RegressionEvaluation, ...]
+
+    model_config = {"frozen": True}
+
+    def assert_passed(self) -> None:
+        if self.passed:
+            return
+        passed_runs = sum(run.passed for run in self.runs)
+        failed = [
+            f"run {index}: {'; '.join(run.failures)}"
+            for index, run in enumerate(self.runs, start=1)
+            if not run.passed
+        ]
+        raise AssertionError(
+            f"Regression case {self.case_id!r} pass rate {self.pass_rate:.0%} "
+            f"({passed_runs}/{len(self.runs)}) is below {self.min_pass_rate:.0%}. "
+            f"Failures: {' | '.join(failed)}"
+        )
+
+
 class RegressionCase(BaseModel):
     """A portable production regression represented as data."""
 
@@ -95,13 +171,105 @@ class RegressionCase(BaseModel):
     description: str = Field(min_length=1)
     input: dict[str, Any]
     expected_outcome: RegressionExpectedOutcome
+    goal: str | None = Field(default=None, min_length=1)
+    persona: str | None = Field(default=None, min_length=1)
+    runs: int = Field(default=1, ge=1, le=100)
+    behavior: RegressionBehavior | None = None
     tags: list[str] = Field(default_factory=list)
     created_at: datetime
 
     model_config = {"extra": "forbid"}
 
     def evaluate(self, result: AgentExecutionResult) -> RegressionEvaluation:
-        """Evaluate only declared deterministic expectations."""
+        """Evaluate deterministic expectations and built-in behavior checks for one run."""
+        failures, scores = self._deterministic_failures(result)
+        return RegressionEvaluation(
+            case_id=self.id,
+            passed=not failures,
+            failures=tuple(failures),
+            scores=scores,
+        )
+
+    async def evaluate_runs(
+        self,
+        results: Sequence[AgentExecutionResult],
+        *,
+        scorers: Iterable[Scorer] = (),
+    ) -> RegressionRunEvaluation:
+        """Evaluate every run, including named scores, against ``min_pass_rate``."""
+        if not results:
+            raise ValueError(f"Regression case {self.id!r} requires at least one result")
+        named_scorers = self._resolve_scorers(scorers)
+        thresholds = self.behavior.scores if self.behavior else {}
+
+        evaluations: list[RegressionEvaluation] = []
+        for result in results:
+            failures, scores = self._deterministic_failures(result)
+            for name, minimum in thresholds.items():
+                score = await evaluate_score(named_scorers[name], result)
+                scores[name] = score.value
+                if score.value < minimum:
+                    detail = f" ({score.reason})" if score.reason else ""
+                    failures.append(
+                        f"Score {name!r}: expected >= {minimum:.2f}. "
+                        f"Observed: {score.value:.2f}{detail}"
+                    )
+            evaluations.append(
+                RegressionEvaluation(
+                    case_id=self.id,
+                    passed=not failures,
+                    failures=tuple(failures),
+                    scores=scores,
+                )
+            )
+
+        min_pass_rate = self.behavior.min_pass_rate if self.behavior else 1.0
+        pass_rate = sum(evaluation.passed for evaluation in evaluations) / len(evaluations)
+        return RegressionRunEvaluation(
+            case_id=self.id,
+            goal=self.goal,
+            persona=self.persona,
+            passed=pass_rate >= min_pass_rate,
+            pass_rate=pass_rate,
+            min_pass_rate=min_pass_rate,
+            runs=tuple(evaluations),
+        )
+
+    async def run(
+        self,
+        executor: AsyncAgentExecutor | AsyncExecutionCallable,
+        *,
+        scorers: Iterable[Scorer] = (),
+        idempotency_key: str | None = None,
+    ) -> RegressionRunEvaluation:
+        """Execute the case ``runs`` times with one idempotency key and evaluate all runs."""
+        scorer_list = list(scorers)
+        self._resolve_scorers(scorer_list)
+        aggregate = await run_repeatedly(
+            executor,
+            input=self.input,
+            runs=self.runs,
+            idempotency_key=idempotency_key or f"regression:{self.id}",
+        )
+        return await self.evaluate_runs(aggregate.results, scorers=scorer_list)
+
+    def _resolve_scorers(self, scorers: Iterable[Scorer]) -> dict[str, Scorer]:
+        named: dict[str, Scorer] = {}
+        for scorer in scorers:
+            if scorer.name in named:
+                raise ValueError(f"Duplicate scorer name {scorer.name!r}")
+            named[scorer.name] = scorer
+        required = self.behavior.scores if self.behavior else {}
+        missing = sorted(set(required) - set(named))
+        if missing:
+            raise ValueError(
+                f"Regression case {self.id!r} declares scores with no matching scorer: {missing}"
+            )
+        return named
+
+    def _deterministic_failures(
+        self, result: AgentExecutionResult
+    ) -> tuple[list[str], dict[str, float]]:
         expected = self.expected_outcome
         engine = FlowAssertionEngine(result)
         failures: list[str] = []
@@ -186,11 +354,14 @@ def evaluate(self, result: AgentExecutionResult) -> RegressionEvaluation:
         if expected.read_only:
             _capture_assertion(failures, "Read-only workflow", engine.assert_read_only)
 
-        return RegressionEvaluation(
-            case_id=self.id,
-            passed=not failures,
-            failures=tuple(failures),
-        )
+        scores: dict[str, float] = {}
+        for scorer in self.behavior.builtin_scorers() if self.behavior else []:
+            score = scorer.score(result)
+            scores[scorer.name] = score.value
+            if score.value < 1.0:
+                failures.append(f"Behavior {scorer.name!r}: {score.reason}")
+
+        return failures, scores
 
 
 def load_regression_cases(path: str | Path) -> list[RegressionCase]:
diff --git a/tests/test_regression_behavior.py b/tests/test_regression_behavior.py
new file mode 100644
index 0000000..bce8624
--- /dev/null
+++ b/tests/test_regression_behavior.py
@@ -0,0 +1,298 @@
+"""Goal-based regression cases evaluated across repeated runs."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from agent_test_kit import (
+    AgentExecutionResult,
+    RegressionCase,
+    RegressionCaseLoadError,
+    RunContext,
+    Score,
+    load_regression_cases,
+)
+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
+
+EXAMPLE_CASES = Path(__file__).parents[1] / "examples" / "behavior_regression_cases.yaml"
+
+
+def _case(**overrides: Any) -> RegressionCase:
+    payload: dict[str, Any] = {
+        "id": "refund-status",
+        "source": "PROD-1",
+        "description": "Refund status lookup",
+        "goal": "User learns refund status",
+        "persona": "frustrated_customer",
+        "runs": 4,
+        "input": {"order_id": 123},
+        "expected_outcome": {"success": True, "read_only": True},
+        "behavior": {
+            "min_pass_rate": 0.75,
+            "required_terms": ["123"],
+            "forbidden_terms": ["api key"],
+        },
+        "created_at": "2026-09-24T00:00:00Z",
+    }
+    payload.update(overrides)
+    return RegressionCase.model_validate(payload)
+
+
+def _result(response: Any, *, write: bool = False, run: int = 1) -> AgentExecutionResult:
+    calls = []
+    if write:
+        calls.append(
+            ToolCall(
+                server="billing",
+                name="issue_refund",
+                operation_kind=ToolOperationKind.WRITE,
+                status=ToolCallStatus.SUCCESS,
+            )
+        )
+    return AgentExecutionResult(
+        success=True,
+        run_id=f"run-{run}",
+        response=response,
+        trace=AgentTrace(tool_calls=calls),
+    )
+
+
+class ScriptedExecutor:
+    def __init__(self, results: list[AgentExecutionResult]) -> None:
+        self._results = iter(results)
+        self.inputs: list[dict[str, Any]] = []
+        self.keys: list[str | None] = []
+
+    async def execute(
+        self,
+        input: dict[str, Any],
+        *,
+        run_context: RunContext | None = None,
+    ) -> AgentExecutionResult:
+        self.inputs.append(input)
+        self.keys.append(run_context.idempotency_key if run_context else None)
+        return next(self._results)
+
+
+class NamedScorer:
+    def __init__(self, name: str, values: list[float]) -> None:
+        self.name = name
+        self._values = iter(values)
+
+    async def score(self, result: AgentExecutionResult) -> Score:
+        return Score(next(self._values), reason=f"{self.name} rationale")
+
+
+def test_example_yaml_loads_goal_cases() -> None:
+    cases = load_regression_cases(EXAMPLE_CASES)
+
+    assert [case.id for case in cases] == [
+        "refund_status_frustrated_customer",
+        "order_lookup_structured",
+    ]
+    first = cases[0]
+    assert first.goal == "User learns the current refund status for order 123"
+    assert first.persona == "frustrated_customer"
+    assert first.runs == 5
+    assert first.behavior is not None
+    assert first.behavior.min_pass_rate == 0.8
+    assert first.behavior.scores == {"groundedness": 0.8}
+
+
+def test_cases_without_new_fields_keep_previous_defaults() -> None:
+    case = _case(goal=None, persona=None, behavior=None, runs=1)
+
+    evaluation = case.evaluate(_result("anything"))
+
+    assert case.runs == 1
+    assert evaluation.passed
+    assert evaluation.scores == {}
+
+
+@pytest.mark.parametrize(
+    ("override", "field"),
+    [
+        ({"runs": 0}, "runs"),
+        ({"runs": 101}, "runs"),
+        ({"goal": ""}, "goal"),
+        ({"behavior": {"min_pass_rate": 0}}, "min_pass_rate"),
+        ({"behavior": {"min_pass_rate": 1.5}}, "min_pass_rate"),
+        ({"behavior": {"scores": {"judge": 1.2}}}, "scores"),
+        ({"behavior": {"unknown": True}}, "unknown"),
+    ],
+)
+def test_behavior_schema_validation(override: dict[str, Any], field: str) -> None:
+    with pytest.raises(ValidationError, match=field):
+        _case(**override)
+
+
+def test_loader_reports_behavior_errors_with_case_context(tmp_path: Path) -> None:
+    path = tmp_path / "bad.yaml"
+    path.write_text(
+        """
+cases:
+  - id: broken-behavior
+    source: PROD-2
+    description: invalid threshold
+    input: {}
+    expected_outcome: {success: true}
+    behavior: {min_pass_rate: 2}
+    created_at: 2026-09-24T00:00:00Z
+""",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(RegressionCaseLoadError, match=r"bad.yaml.*broken-behavior.*min_pass_rate"):
+        load_regression_cases(path)
+
+
+@pytest.mark.parametrize(
+    ("response", "expected_failure", "scores"),
+    [
+        (
+            "Refund for 123 is pending",
+            None,
+            {"required_terms": 1.0, "forbidden_terms": 1.0},
+        ),
+        (
+            "Your refund is pending",
+            "Behavior 'required_terms': Missing terms: ['123']",
+            {"required_terms": 0.0, "forbidden_terms": 1.0},
+        ),
+        (
+            "Order 123, api key abc",
+            "Behavior 'forbidden_terms': Forbidden terms present: ['api key']",
+            {"required_terms": 1.0, "forbidden_terms": 0.0},
+        ),
+    ],
+)
+def test_evaluate_applies_builtin_behavior_checks(
+    response: str, expected_failure: str | None, scores: dict[str, float]
+) -> None:
+    evaluation = _case().evaluate(_result(response))
+
+    assert evaluation.scores == scores
+    if expected_failure is None:
+        assert evaluation.passed
+    else:
+        assert evaluation.failures == (expected_failure,)
+
+
+def test_evaluate_applies_required_fields() -> None:
+    case = _case(behavior={"required_fields": ["order.status", "order.delivery_date"]})
+
+    evaluation = case.evaluate(_result({"order": {"status": "shipped"}}))
+
+    assert not evaluation.passed
+    assert evaluation.scores == {"required_fields": 0.5}
+    assert "order.delivery_date" in evaluation.failures[0]
+
+
+@pytest.mark.asyncio
+async def test_run_passes_when_variance_stays_within_pass_rate() -> None:
+    executor = ScriptedExecutor(
+        [
+            _result("123 pending", run=1),
+            _result("pending, order 123", run=2),
+            _result("cannot find it", run=3),
+            _result("123 approved", run=4),
+        ]
+    )
+
+    evaluation = await _case().run(executor)
+
+    assert evaluation.passed
+    assert evaluation.pass_rate == 0.75
+    assert evaluation.min_pass_rate == 0.75
+    assert evaluation.goal == "User learns refund status"
+    assert evaluation.persona == "frustrated_customer"
+    assert [run.passed for run in evaluation.runs] == [True, True, False, True]
+    assert executor.inputs == [{"order_id": 123}] * 4
+    assert set(executor.keys) == {"regression:refund-status"}
+    evaluation.assert_passed()
+
+
+@pytest.mark.asyncio
+async def test_run_fails_below_pass_rate_with_per_run_failures() -> None:
+    executor = ScriptedExecutor(
+        [
+            _result("123 pending", run=1),
+            _result("pending", run=2),
+            _result("123", write=True, run=3),
+            _result("123 ok", run=4),
+        ]
+    )
+
+    evaluation = await _case().run(executor, idempotency_key="custom-key")
+
+    assert not evaluation.passed
+    assert evaluation.pass_rate == 0.5
+    assert set(executor.keys) == {"custom-key"}
+    with pytest.raises(AssertionError) as exc_info:
+        evaluation.assert_passed()
+    message = str(exc_info.value)
+    assert "pass rate 50% (2/4) is below 75%" in message
+    assert "run 2: Behavior 'required_terms'" in message
+    assert "run 3: Read-only workflow" in message
+
+
+@pytest.mark.asyncio
+async def test_evaluate_runs_applies_named_scores() -> None:
+    case = _case(behavior={"min_pass_rate": 0.5, "scores": {"groundedness": 0.8}})
+    results = [_result("a", run=1), _result("b", run=2)]
+
+    evaluation = await case.evaluate_runs(
+        results, scorers=[NamedScorer("groundedness", [0.9, 0.4]), NamedScorer("unused", [])]
+    )
+
+    assert evaluation.passed
+    assert [run.scores for run in evaluation.runs] == [
+        {"groundedness": 0.9},
+        {"groundedness": 0.4},
+    ]
+    assert evaluation.runs[1].failures == (
+        "Score 'groundedness': expected >= 0.80. Observed: 0.40 (groundedness rationale)",
+    )
+
+
+@pytest.mark.asyncio
+async def test_run_rejects_missing_scorer_before_executing() -> None:
+    case = _case(behavior={"scores": {"groundedness": 0.8}})
+    executor = ScriptedExecutor([])
+
+    with pytest.raises(ValueError, match=r"no matching scorer: \['groundedness'\]"):
+        await case.run(executor)
+    assert executor.inputs == []
+
+
+@pytest.mark.asyncio
+async def test_evaluate_runs_rejects_duplicate_scorer_names() -> None:
+    with pytest.raises(ValueError, match="Duplicate scorer name 'judge'"):
+        await _case().evaluate_runs(
+            [_result("123")], scorers=[NamedScorer("judge", []), NamedScorer("judge", [])]
+        )
+
+
+@pytest.mark.asyncio
+async def test_evaluate_runs_requires_results() -> None:
+    with pytest.raises(ValueError, match="at least one result"):
+        await _case().evaluate_runs([])
+
+
+@pytest.mark.asyncio
+async def test_case_without_behavior_requires_every_run_to_pass() -> None:
+    case = _case(behavior=None)
+
+    evaluation = await case.evaluate_runs([_result("ok"), _result("x", write=True)])
+
+    assert evaluation.min_pass_rate == 1.0
+    assert evaluation.pass_rate == 0.5
+    assert not evaluation.passed

From dc8de3ed8fc2cc873c4fad65177d6df546a7f61a Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 17:42:32 +0000
Subject: [PATCH 4/6] Document goal-based regression cases

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 README.md | 40 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git a/README.md b/README.md
index b367de3..49e7d70 100644
--- a/README.md
+++ b/README.md
@@ -381,6 +381,45 @@ for case in cases:
 
 Mark regression tests with `@pytest.mark.agent_regression`.
 
+**Goal-based cases.** Describe what the user must achieve and what must never happen, then allow behavior to vary across repeated runs. All fields are optional; existing cases are unchanged.
+
+| Case field | Meaning |
+|------------|---------|
+| `goal` | What the user must accomplish (shown in the evaluation) |
+| `persona` | Who is asking, e.g. `frustrated_customer` |
+| `runs` | How many times `case.run()` executes the input (default `1`) |
+| `behavior.min_pass_rate` | Fraction of runs that must pass (default `1.0`) |
+| `behavior.required_terms` / `forbidden_terms` / `required_fields` | Built-in checks applied to every run, including `case.evaluate()` |
+| `behavior.scores` | `scorer_name: min_score` for scorers you pass to `case.run()` |
+
+```yaml
+- id: refund_status_frustrated_customer
+  source: PROD-412
+  description: Customer asks angrily where their refund is
+  goal: User learns the current refund status for order 123
+  persona: frustrated_customer
+  runs: 5
+  input: {message: "Where is my refund for order 123?"}
+  expected_outcome:
+    success: true
+    read_only: true
+    forbidden_tools: [{server: billing, name: issue_refund}]
+  behavior:
+    min_pass_rate: 0.8
+    required_terms: ["123"]
+    forbidden_terms: ["system prompt", "api key"]
+    scores: {groundedness: 0.8}
+  created_at: 2026-09-24T00:00:00Z
+```
+
+```python
+for case in load_regression_cases("tests/data/behavior_cases.yaml"):
+    evaluation = await case.run(agent_client, scorers=[GroundednessJudge()])
+    evaluation.assert_passed()  # fails with pass rate and per-run failures
+```
+
+`case.run()` executes `runs` times with one idempotency key (`regression:<id>` 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)
 
 ```python
@@ -672,6 +711,7 @@ from agent_test_kit import (
     JsonReportWriter,
     OptionalStep,
     PromptReviewClient,
+    RegressionRunEvaluation,
     RepeatedExecutionResult,
     RequiredFieldsScorer,
     RequiredTermsScorer,

From d48b4a96beb8204183f97ea964b8d08e8f3d0973 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 18:08:20 +0000
Subject: [PATCH 5/6] Add release readiness report with persona segments,
 baseline comparison, and opt-in Go/No-Go enforcement

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 src/agent_test_kit/__init__.py              |   4 +
 src/agent_test_kit/client/config.py         |   4 +
 src/agent_test_kit/models/report.py         |  40 ++
 src/agent_test_kit/plugin/pytest_plugin.py  |  54 ++-
 src/agent_test_kit/reporting/html_report.py |  79 ++++
 src/agent_test_kit/reporting/json_report.py |  14 +-
 src/agent_test_kit/reporting/readiness.py   | 208 +++++++++
 src/agent_test_kit_pytest_entrypoint.py     |  19 +
 tests/test_pytest_plugin.py                 |   5 +-
 tests/test_readiness.py                     | 448 ++++++++++++++++++++
 10 files changed, 871 insertions(+), 4 deletions(-)
 create mode 100644 src/agent_test_kit/reporting/readiness.py
 create mode 100644 tests/test_readiness.py

diff --git a/src/agent_test_kit/__init__.py b/src/agent_test_kit/__init__.py
index b7591b8..1e475cf 100644
--- a/src/agent_test_kit/__init__.py
+++ b/src/agent_test_kit/__init__.py
@@ -40,6 +40,7 @@
 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.reporting.readiness import ReadinessPolicy, evaluate_readiness, load_report
 from agent_test_kit.scoring import (
     ForbiddenTermsScorer,
     RequiredFieldsScorer,
@@ -76,6 +77,7 @@
     "PromptReviewClient",
     "PromptAiHelperProfile",
     "PublishResult",
+    "ReadinessPolicy",
     "RegressionBehavior",
     "RegressionCase",
     "RegressionCaseLoadError",
@@ -98,7 +100,9 @@
     "VerificationResult",
     "VerificationSummary",
     "__version__",
+    "evaluate_readiness",
     "load_regression_cases",
+    "load_report",
     "get_profile",
     "response_text",
     "run_repeatedly",
diff --git a/src/agent_test_kit/client/config.py b/src/agent_test_kit/client/config.py
index 05e9b56..07ddc1a 100644
--- a/src/agent_test_kit/client/config.py
+++ b/src/agent_test_kit/client/config.py
@@ -31,6 +31,10 @@ class AgentTestConfig(BaseSettings):
     environment: str = Field(default="local")
     model: str | None = None
     prompt_version: str | None = None
+    knowledge_version: str | None = Field(
+        default=None,
+        description="Version of the RAG index or knowledge base the agent answered from",
+    )
     repository: str | None = None
     branch: str | None = None
     commit: str | None = None
diff --git a/src/agent_test_kit/models/report.py b/src/agent_test_kit/models/report.py
index 5521b27..688f0a4 100644
--- a/src/agent_test_kit/models/report.py
+++ b/src/agent_test_kit/models/report.py
@@ -70,6 +70,12 @@ class ScenarioReport(BaseModel):
     golden_outcome: str | None = None
     expected_classification: str | None = None
     golden_met: bool | None = None
+    case_id: str | None = None
+    goal: str | None = None
+    persona: str | None = None
+    pass_rate: float | None = None
+    min_pass_rate: float | None = None
+    run_count: int | None = None
 
     @model_validator(mode="after")
     def derive_status(self) -> ScenarioReport:
@@ -97,6 +103,38 @@ class RunMetrics(BaseModel):
     security_failure_count: int = 0
 
 
+class PersonaSegment(BaseModel):
+    """Pass rate of executed scenarios that share one persona."""
+
+    persona: str
+    scenarios: int
+    passed: int
+    pass_rate: float
+
+
+class BaselineComparison(BaseModel):
+    """Differences between this run and a previous report."""
+
+    baseline_run_id: str
+    regressions: list[str] = Field(default_factory=list)
+    pass_rate_drops: list[str] = Field(default_factory=list)
+    missing_scenarios: list[str] = Field(default_factory=list)
+    fingerprint_changes: list[str] = Field(default_factory=list)
+
+
+class ReleaseReadiness(BaseModel):
+    """Go / No-Go summary derived from the run's evidence.
+
+    ``blocking`` reasons produce ``no_go``; ``notes`` inform the reviewer without blocking.
+    """
+
+    verdict: Literal["go", "no_go"]
+    blocking: list[str] = Field(default_factory=list)
+    notes: list[str] = Field(default_factory=list)
+    persona_segments: list[PersonaSegment] = Field(default_factory=list)
+    baseline: BaselineComparison | None = None
+
+
 class AgentTestRunReport(BaseModel):
     """Versioned report emitted after a pytest session."""
 
@@ -111,6 +149,7 @@ class AgentTestRunReport(BaseModel):
     agent_version: str | None = None
     model: str | None = None
     prompt_version: str | None = None
+    knowledge_version: str | None = None
     environment: str
     started_at: datetime
     completed_at: datetime
@@ -118,6 +157,7 @@ class AgentTestRunReport(BaseModel):
     metrics: RunMetrics = Field(default_factory=RunMetrics)
     endpoint_profile: dict[str, str] | None = None
     expected_mcp_servers: list[str] = Field(default_factory=list)
+    readiness: ReleaseReadiness | None = None
 
     def finalize_metrics(self) -> None:
         self.metrics.total_scenarios = len(self.scenarios)
diff --git a/src/agent_test_kit/plugin/pytest_plugin.py b/src/agent_test_kit/plugin/pytest_plugin.py
index 5e37ea5..4a400d0 100644
--- a/src/agent_test_kit/plugin/pytest_plugin.py
+++ b/src/agent_test_kit/plugin/pytest_plugin.py
@@ -6,7 +6,7 @@
 from collections.abc import AsyncIterator, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import pytest
 import pytest_asyncio
@@ -17,9 +17,10 @@
 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.models.execution import AgentExecutionResult
-from agent_test_kit.models.report import ScenarioStatus
+from agent_test_kit.models.report import AgentTestRunReport, ReleaseReadiness, ScenarioStatus
 from agent_test_kit.reporting.html_report import HtmlReportWriter
 from agent_test_kit.reporting.json_report import JsonReportWriter
+from agent_test_kit.reporting.readiness import ReadinessPolicy, evaluate_readiness, load_report
 from agent_test_kit.verifiers import (
     SideEffectVerifier,
     VerificationContext,
@@ -28,6 +29,9 @@
     run_verifiers,
 )
 
+if TYPE_CHECKING:
+    from agent_test_kit.regression import RegressionRunEvaluation
+
 
 @dataclass
 class PluginState:
@@ -35,9 +39,13 @@ class PluginState:
     json_writer: JsonReportWriter | None
     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)
     phase_reports: dict[str, dict[str, pytest.TestReport]] = field(default_factory=dict)
     html_path: Path | None = None
     write_json: bool = True
+    baseline_path: Path | None = None
+    readiness_policy: ReadinessPolicy = field(default_factory=ReadinessPolicy)
+    enforce_readiness: bool = False
 
 
 _STASH_KEY = pytest.StashKey[PluginState]()
@@ -72,11 +80,17 @@ def pytest_configure(config: pytest.Config) -> None:
         profile = get_profile(profile_env)
         writer.endpoint_profile = dict(profile.endpoint_rows())
         writer.expected_mcp_servers = list(profile.mcp_servers)
+    baseline_path = config.getoption("--agent-baseline", default=None)
     config.stash[_STASH_KEY] = PluginState(
         agent_config=agent_config,
         json_writer=writer,
         html_path=Path(html_path) if html_path else None,
         write_json=bool(json_path),
+        baseline_path=Path(baseline_path) if baseline_path else None,
+        readiness_policy=ReadinessPolicy(
+            min_persona_pass_rate=config.getoption("--agent-min-persona-pass-rate", default=None)
+        ),
+        enforce_readiness=bool(config.getoption("--agent-enforce-readiness", default=False)),
     )
 
 
@@ -101,6 +115,12 @@ def attach_verifier_results(
     ) -> None:
         attach_verification_results(self.config, self.nodeid, results)
 
+    def attach_regression_evaluation(self, evaluation: RegressionRunEvaluation) -> None:
+        """Record goal, persona, and pass rate for readiness segmentation."""
+        state = self.config.stash.get(_STASH_KEY, None)
+        if state is not None:
+            state.regression_evaluations[self.nodeid] = evaluation
+
     async def verify(
         self,
         verifiers: Sequence[SideEffectVerifier],
@@ -224,6 +244,7 @@ def _record_scenario_from_reports(
         extra=extra,
         execution_result=stored_result,
         verification_results=state.verification_results.get(item.nodeid, []),
+        regression_evaluation=state.regression_evaluations.get(item.nodeid),
     )
 
 
@@ -248,10 +269,39 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
     if state is None or state.json_writer is None:
         return
     report = state.json_writer.build_report()
+    report.readiness = _evaluate_session_readiness(state, report)
     if state.write_json:
         state.json_writer.write(report)
     if state.html_path is not None:
         HtmlReportWriter(state.html_path).write(report)
+    if (
+        state.enforce_readiness
+        and report.readiness.verdict == "no_go"
+        and exitstatus == pytest.ExitCode.OK
+    ):
+        session.exitstatus = pytest.ExitCode.TESTS_FAILED
+
+
+def _evaluate_session_readiness(
+    state: PluginState,
+    report: AgentTestRunReport,
+) -> ReleaseReadiness:
+    baseline: AgentTestRunReport | None = None
+    notes: list[str] = []
+    if state.baseline_path is not None:
+        try:
+            baseline = load_report(state.baseline_path)
+        except FileNotFoundError:
+            notes.append(f"Baseline report not found: {state.baseline_path}")
+        except (OSError, ValueError) as exc:
+            path, reason = state.baseline_path, type(exc).__name__
+            notes.append(f"Baseline report unreadable: {path} ({reason})")
+    return evaluate_readiness(
+        report,
+        baseline=baseline,
+        policy=state.readiness_policy,
+        notes=notes,
+    )
 
 
 def store_execution_result(
diff --git a/src/agent_test_kit/reporting/html_report.py b/src/agent_test_kit/reporting/html_report.py
index 9bf2cdf..592c4e6 100644
--- a/src/agent_test_kit/reporting/html_report.py
+++ b/src/agent_test_kit/reporting/html_report.py
@@ -164,6 +164,76 @@ def _golden_outcome_block(scenario: dict[str, Any]) -> str:
 </div>"""
 
 
+def _bullet_list(items: list[Any]) -> str:
+    return "<ul>" + "".join(f"<li>{_text(item)}</li>" for item in items) + "</ul>"
+
+
+def _readiness_block(readiness: dict[str, Any] | None) -> str:
+    if not readiness:
+        return ""
+    go = readiness.get("verdict") == "go"
+    css = "readiness-go" if go else "readiness-no-go"
+    label = "GO" if go else "NO-GO"
+    parts = [
+        f'<section class="card readiness {css}">'
+        f'<h2>Release readiness <span class="verdict">{label}</span></h2>'
+        '<p class="readiness-hint">Evidence for a human release decision.</p>'
+    ]
+    blocking = readiness.get("blocking") or []
+    if blocking:
+        parts.append(f"<h4>Blocking</h4>{_bullet_list(blocking)}")
+    notes = readiness.get("notes") or []
+    if notes:
+        parts.append(f"<h4>Notes</h4>{_bullet_list(notes)}")
+    segments = readiness.get("persona_segments") or []
+    if segments:
+        rows = [
+            [
+                segment["persona"],
+                segment["scenarios"],
+                segment["passed"],
+                f"{segment['pass_rate']:.0%}",
+            ]
+            for segment in segments
+        ]
+        parts.append(
+            "<h4>Personas</h4>" + _table(["Persona", "Scenarios", "Passed", "Pass rate"], rows)
+        )
+    baseline = readiness.get("baseline")
+    if baseline:
+        rows = [
+            [change, values]
+            for change, values in (
+                ("Regressions", baseline.get("regressions")),
+                ("Pass-rate drops", baseline.get("pass_rate_drops")),
+                ("Not run", baseline.get("missing_scenarios")),
+                ("Fingerprint changes", baseline.get("fingerprint_changes")),
+            )
+            if values
+        ]
+        body = _table(["Change", "Details"], rows) if rows else "<p>No changes.</p>"
+        parts.append(f"<h4>Baseline {_text(baseline.get('baseline_run_id'))}</h4>{body}")
+    parts.append("</section>")
+    return "".join(parts)
+
+
+def _behavior_rows(scenario: dict[str, Any]) -> list[list[Any]]:
+    rows: list[list[Any]] = []
+    if scenario.get("case_id"):
+        rows.append(["Case", scenario["case_id"]])
+    if scenario.get("goal"):
+        rows.append(["Goal", scenario["goal"]])
+    if scenario.get("persona"):
+        rows.append(["Persona", scenario["persona"]])
+    if scenario.get("pass_rate") is not None:
+        required = scenario.get("min_pass_rate")
+        suffix = f" (required {required:.0%})" if required is not None else ""
+        rows.append(["Pass rate", f"{scenario['pass_rate']:.0%}{suffix}"])
+    if scenario.get("run_count") is not None:
+        rows.append(["Runs", scenario["run_count"]])
+    return rows
+
+
 class HtmlReportWriter:
     """Render an escaped and redacted report as one portable HTML file."""
 
@@ -180,6 +250,7 @@ def write(self, report: AgentTestRunReport) -> Path:
             ["Agent version", payload.get("agent_version")],
             ["Model", payload.get("model")],
             ["Prompt version", payload.get("prompt_version")],
+            ["Knowledge version", payload.get("knowledge_version")],
             ["Environment", payload["environment"]],
             ["Repository", payload.get("repository")],
             ["Branch / commit", f"{payload.get('branch') or ''} {payload.get('commit') or ''}"],
@@ -265,6 +336,12 @@ def write(self, report: AgentTestRunReport) -> Path:
 .classification-chip{{font-size:.7rem;font-weight:700;padding:.15rem .45rem;border-radius:4px;background:#eef4ff;color:var(--accent)}}
 .violation-chips{{display:flex;flex-wrap:wrap;gap:.35rem;margin-top:.4rem}}
 .violation-chip{{font:10px ui-monospace;padding:.15rem .4rem;border-radius:4px;background:#f2f4f7;color:var(--muted)}}
+.readiness{{border-left:6px solid var(--line)}}
+.readiness-go{{border-left-color:var(--pass)}}.readiness-no-go{{border-left-color:var(--fail)}}
+.verdict{{font-size:.85rem;font-weight:800;padding:.2rem .6rem;border-radius:6px;color:#fff}}
+.verdict{{margin-left:.5rem;vertical-align:middle}}
+.readiness-go .verdict{{background:var(--pass)}}.readiness-no-go .verdict{{background:var(--fail)}}
+.readiness-hint{{color:var(--muted);margin:.2rem 0 .5rem}}
 </style>
 </head>
 <body><main>
@@ -273,6 +350,7 @@ def write(self, report: AgentTestRunReport) -> Path:
 <div class="hero-meta">Framework {escape(str(payload.get("framework_version") or ""))} · Run {escape(str(payload["run_id"]))}</div>
 <div class="badges"><span class="badge">Agent: {agent_badge}</span><span class="badge">Env: {env_badge}</span></div>
 </header>
+{_readiness_block(payload.get("readiness"))}
 {_metric_cards(metrics)}
 {endpoint_html}
 {mcp_overview}
@@ -329,6 +407,7 @@ def _scenario(self, scenario: dict[str, Any]) -> str:
             for event in scenario["timeline"]
         ]
         summary = [
+            *_behavior_rows(scenario),
             ["Node", scenario["nodeid"]],
             ["Duration (ms)", scenario["duration_ms"]],
             ["Run", scenario.get("run_id")],
diff --git a/src/agent_test_kit/reporting/json_report.py b/src/agent_test_kit/reporting/json_report.py
index dd93474..a508e88 100644
--- a/src/agent_test_kit/reporting/json_report.py
+++ b/src/agent_test_kit/reporting/json_report.py
@@ -7,7 +7,7 @@
 from collections.abc import Sequence
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from agent_test_kit.client.config import AgentTestConfig
 from agent_test_kit.models.execution import AgentExecutionResult
@@ -22,6 +22,9 @@
 from agent_test_kit.reporting.redaction import redact_value
 from agent_test_kit.verifiers.protocols import VerificationResult
 
+if TYPE_CHECKING:
+    from agent_test_kit.regression import RegressionRunEvaluation
+
 
 def _evidence_links(value: Any) -> list[str]:
     if isinstance(value, str):
@@ -64,8 +67,10 @@ def record_scenario(
         extra: dict[str, Any] | None = None,
         execution_result: AgentExecutionResult | None = None,
         verification_results: Sequence[VerificationResult] = (),
+        regression_evaluation: RegressionRunEvaluation | None = None,
     ) -> None:
         extra = extra or {}
+        evaluation = regression_evaluation
         if execution_result is None:
             candidate = extra.get("execution_result")
             if isinstance(candidate, AgentExecutionResult):
@@ -189,6 +194,12 @@ def record_scenario(
                 golden_met=extra.get("golden_met")
                 if extra.get("golden_met") is not None
                 else response_dict.get("golden_met"),
+                case_id=evaluation.case_id if evaluation is not None else None,
+                goal=evaluation.goal if evaluation is not None else None,
+                persona=evaluation.persona if evaluation is not None else None,
+                pass_rate=evaluation.pass_rate if evaluation is not None else None,
+                min_pass_rate=evaluation.min_pass_rate if evaluation is not None else None,
+                run_count=len(evaluation.runs) if evaluation is not None else None,
             )
         )
 
@@ -203,6 +214,7 @@ def build_report(self) -> AgentTestRunReport:
             agent_version=self.config.agent_version,
             model=self.config.model,
             prompt_version=self.config.prompt_version,
+            knowledge_version=self.config.knowledge_version,
             environment=self.config.environment,
             started_at=self.started_at,
             completed_at=datetime.now(UTC),
diff --git a/src/agent_test_kit/reporting/readiness.py b/src/agent_test_kit/reporting/readiness.py
new file mode 100644
index 0000000..2324d68
--- /dev/null
+++ b/src/agent_test_kit/reporting/readiness.py
@@ -0,0 +1,208 @@
+"""Go / No-Go release readiness derived from a run report and an optional baseline."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+
+from agent_test_kit.models.report import (
+    AgentTestRunReport,
+    BaselineComparison,
+    PersonaSegment,
+    ReleaseReadiness,
+    ScenarioReport,
+)
+from agent_test_kit.scoring import validate_unit_interval
+
+_FINGERPRINT_FIELDS = ("model", "prompt_version", "agent_version", "knowledge_version")
+_TOKEN_INCREASE_NOTE_THRESHOLD = 0.2
+
+
+@dataclass(frozen=True, slots=True)
+class ReadinessPolicy:
+    """Optional thresholds that turn readiness observations into blocking reasons.
+
+    Failed, errored, security-failing, and duplicate-write scenarios always block.
+    """
+
+    min_persona_pass_rate: float | None = None
+    max_pass_rate_drop: float | None = None
+
+    def __post_init__(self) -> None:
+        if self.min_persona_pass_rate is not None:
+            validate_unit_interval("min_persona_pass_rate", self.min_persona_pass_rate)
+        if self.max_pass_rate_drop is not None:
+            validate_unit_interval("max_pass_rate_drop", self.max_pass_rate_drop)
+
+
+def load_report(path: Path) -> AgentTestRunReport:
+    """Load a JSON report written by ``--agent-report-json``."""
+    return AgentTestRunReport.model_validate_json(path.read_text(encoding="utf-8"))
+
+
+def evaluate_readiness(
+    report: AgentTestRunReport,
+    *,
+    baseline: AgentTestRunReport | None = None,
+    policy: ReadinessPolicy | None = None,
+    notes: Iterable[str] = (),
+) -> ReleaseReadiness:
+    """Summarize whether the run's evidence supports releasing the agent.
+
+    The verdict is ``no_go`` when any blocking reason exists. ``notes`` are informational
+    and never change the verdict.
+    """
+    policy = policy or ReadinessPolicy()
+    executed = [scenario for scenario in report.scenarios if scenario.status != "skipped"]
+    blocking: list[str] = []
+    all_notes = list(notes)
+
+    if not executed:
+        blocking.append("No scenarios executed")
+    failed = [scenario for scenario in executed if scenario.status in {"failed", "error"}]
+    if failed:
+        blocking.append(f"{len(failed)} scenario(s) failed or errored")
+    security_failures = sum(1 for scenario in executed if scenario.security_failure)
+    if security_failures:
+        blocking.append(f"{security_failures} security failure(s)")
+    duplicate_writes = sum(scenario.duplicate_write_count for scenario in executed)
+    if duplicate_writes:
+        blocking.append(f"{duplicate_writes} duplicate write(s)")
+
+    segments = persona_segments(executed)
+    if policy.min_persona_pass_rate is not None:
+        blocking.extend(
+            f"Persona {segment.persona!r} pass rate {segment.pass_rate:.0%} "
+            f"is below {policy.min_persona_pass_rate:.0%}"
+            for segment in segments
+            if segment.pass_rate < policy.min_persona_pass_rate
+        )
+
+    comparison = None
+    if baseline is not None:
+        comparison = compare_to_baseline(report, baseline)
+        if comparison.regressions:
+            blocking.append(
+                f"{len(comparison.regressions)} scenario(s) regressed since baseline "
+                f"{comparison.baseline_run_id}"
+            )
+        if policy.max_pass_rate_drop is not None:
+            blocking.extend(_excessive_drops(report, baseline, policy.max_pass_rate_drop))
+        all_notes.extend(comparison.fingerprint_changes)
+        all_notes.extend(f"Pass rate dropped: {drop}" for drop in comparison.pass_rate_drops)
+        if comparison.missing_scenarios:
+            all_notes.append(
+                f"{len(comparison.missing_scenarios)} baseline scenario(s) were not run"
+            )
+        token_note = _token_usage_note(report, baseline)
+        if token_note:
+            all_notes.append(token_note)
+
+    return ReleaseReadiness(
+        verdict="no_go" if blocking else "go",
+        blocking=blocking,
+        notes=all_notes,
+        persona_segments=segments,
+        baseline=comparison,
+    )
+
+
+def persona_segments(scenarios: Iterable[ScenarioReport]) -> list[PersonaSegment]:
+    """Group executed scenarios by persona, weakest segment first."""
+    grouped: dict[str, list[ScenarioReport]] = {}
+    for scenario in scenarios:
+        if scenario.persona and scenario.status != "skipped":
+            grouped.setdefault(scenario.persona, []).append(scenario)
+    segments = [
+        PersonaSegment(
+            persona=persona,
+            scenarios=len(members),
+            passed=sum(1 for member in members if member.status == "passed"),
+            pass_rate=sum(_scenario_rate(member) for member in members) / len(members),
+        )
+        for persona, members in grouped.items()
+    ]
+    return sorted(segments, key=lambda segment: (segment.pass_rate, segment.persona))
+
+
+def compare_to_baseline(
+    report: AgentTestRunReport,
+    baseline: AgentTestRunReport,
+) -> BaselineComparison:
+    """List scenario regressions, pass-rate drops, and fingerprint changes."""
+    current = {scenario.nodeid: scenario for scenario in report.scenarios}
+    previous = {scenario.nodeid: scenario for scenario in baseline.scenarios}
+    regressions = [
+        nodeid
+        for nodeid, before in previous.items()
+        if before.status == "passed"
+        and nodeid in current
+        and current[nodeid].status in {"failed", "error"}
+    ]
+    drops = [
+        f"{nodeid}: {before:.0%} -> {after:.0%}"
+        for nodeid, before, after in _pass_rate_pairs(current, previous)
+        if after < before
+    ]
+    missing = [
+        nodeid
+        for nodeid, before in previous.items()
+        if before.status != "skipped" and nodeid not in current
+    ]
+    fingerprint_changes = [
+        f"{field} changed: {getattr(baseline, field)} -> {getattr(report, field)}"
+        for field in _FINGERPRINT_FIELDS
+        if getattr(baseline, field) != getattr(report, field)
+    ]
+    return BaselineComparison(
+        baseline_run_id=baseline.run_id,
+        regressions=regressions,
+        pass_rate_drops=drops,
+        missing_scenarios=missing,
+        fingerprint_changes=fingerprint_changes,
+    )
+
+
+def _scenario_rate(scenario: ScenarioReport) -> float:
+    if scenario.pass_rate is not None:
+        return scenario.pass_rate
+    return 1.0 if scenario.status == "passed" else 0.0
+
+
+def _pass_rate_pairs(
+    current: dict[str, ScenarioReport],
+    previous: dict[str, ScenarioReport],
+) -> list[tuple[str, float, float]]:
+    return [
+        (nodeid, _scenario_rate(before), _scenario_rate(current[nodeid]))
+        for nodeid, before in previous.items()
+        if nodeid in current
+        and before.pass_rate is not None
+        and current[nodeid].pass_rate is not None
+    ]
+
+
+def _excessive_drops(
+    report: AgentTestRunReport,
+    baseline: AgentTestRunReport,
+    max_drop: float,
+) -> list[str]:
+    current = {scenario.nodeid: scenario for scenario in report.scenarios}
+    previous = {scenario.nodeid: scenario for scenario in baseline.scenarios}
+    return [
+        f"{nodeid} pass rate dropped {before - after:.0%} (limit {max_drop:.0%})"
+        for nodeid, before, after in _pass_rate_pairs(current, previous)
+        if before - after > max_drop
+    ]
+
+
+def _token_usage_note(report: AgentTestRunReport, baseline: AgentTestRunReport) -> str | None:
+    before = baseline.metrics.total_token_usage
+    after = report.metrics.total_token_usage
+    if not before or after is None:
+        return None
+    increase = (after - before) / before
+    if increase <= _TOKEN_INCREASE_NOTE_THRESHOLD:
+        return None
+    return f"Token usage increased {increase:.0%} ({before} -> {after})"
diff --git a/src/agent_test_kit_pytest_entrypoint.py b/src/agent_test_kit_pytest_entrypoint.py
index 7a12277..9a8963e 100644
--- a/src/agent_test_kit_pytest_entrypoint.py
+++ b/src/agent_test_kit_pytest_entrypoint.py
@@ -20,6 +20,25 @@ def pytest_addoption(parser: pytest.Parser) -> None:
         default=None,
         help="Write self-contained agent test HTML report to this path",
     )
+    group.addoption(
+        "--agent-baseline",
+        action="store",
+        default=None,
+        help="Compare readiness against a previous --agent-report-json file",
+    )
+    group.addoption(
+        "--agent-min-persona-pass-rate",
+        action="store",
+        type=float,
+        default=None,
+        help="Block readiness when any persona pass rate is below this value (0.0-1.0)",
+    )
+    group.addoption(
+        "--agent-enforce-readiness",
+        action="store_true",
+        default=False,
+        help="Fail the pytest session when release readiness is No-Go",
+    )
 
 
 @pytest.hookimpl(trylast=True)
diff --git a/tests/test_pytest_plugin.py b/tests/test_pytest_plugin.py
index be21ed9..6660d83 100644
--- a/tests/test_pytest_plugin.py
+++ b/tests/test_pytest_plugin.py
@@ -16,8 +16,9 @@
 
 @pytest.mark.agent_unit
 def test_pytest_configure_creates_json_writer(tmp_path: Path) -> None:
+    options = {"--agent-report-json": str(tmp_path / "report.json")}
     config = MagicMock()
-    config.getoption.return_value = str(tmp_path / "report.json")
+    config.getoption.side_effect = lambda name, default=None: options.get(name, default)
     config.stash = pytest.Stash()
 
     pytest_plugin.pytest_configure(config)
@@ -25,6 +26,8 @@ def test_pytest_configure_creates_json_writer(tmp_path: Path) -> None:
     state = config.stash[pytest_plugin._STASH_KEY]
     assert isinstance(state.agent_config, AgentTestConfig)
     assert state.json_writer is not None
+    assert state.baseline_path is None
+    assert state.enforce_readiness is False
 
 
 @pytest.mark.agent_unit
diff --git a/tests/test_readiness.py b/tests/test_readiness.py
new file mode 100644
index 0000000..92f2ca4
--- /dev/null
+++ b/tests/test_readiness.py
@@ -0,0 +1,448 @@
+"""Release readiness: persona segments, baseline comparison, and Go / No-Go."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from agent_test_kit import AgentTestConfig, ReadinessPolicy, evaluate_readiness, load_report
+from agent_test_kit.models.report import AgentTestRunReport, ScenarioReport, ScenarioStatus
+from agent_test_kit.plugin import pytest_plugin
+from agent_test_kit.regression import RegressionEvaluation, RegressionRunEvaluation
+from agent_test_kit.reporting.html_report import HtmlReportWriter
+from agent_test_kit.reporting.json_report import JsonReportWriter
+
+pytestmark = pytest.mark.agent_unit
+
+_NOW = datetime(2026, 9, 24, tzinfo=UTC)
+
+
+def _scenario(
+    nodeid: str,
+    status: ScenarioStatus = "passed",
+    **fields: Any,
+) -> ScenarioReport:
+    return ScenarioReport(
+        name=nodeid.rsplit("::", 1)[-1],
+        nodeid=nodeid,
+        passed=status == "passed",
+        status=status,
+        duration_ms=1,
+        **fields,
+    )
+
+
+def _report(
+    *scenarios: ScenarioReport, run_id: str = "run_now", **fields: Any
+) -> AgentTestRunReport:
+    report = AgentTestRunReport(
+        run_id=run_id,
+        agent_id="support-agent",
+        environment="ci",
+        started_at=_NOW,
+        completed_at=_NOW,
+        scenarios=list(scenarios),
+        **fields,
+    )
+    report.finalize_metrics()
+    return report
+
+
+def _evaluation(
+    persona: str | None = "frustrated_customer",
+    *,
+    pass_rate: float = 1.0,
+    runs: int = 1,
+) -> RegressionRunEvaluation:
+    return RegressionRunEvaluation(
+        case_id="refund_status",
+        goal="Customer learns refund status",
+        persona=persona,
+        passed=pass_rate >= 0.6,
+        pass_rate=pass_rate,
+        min_pass_rate=0.6,
+        runs=tuple(RegressionEvaluation(case_id="refund_status", passed=True) for _ in range(runs)),
+    )
+
+
+class TestReadinessPolicy:
+    @pytest.mark.parametrize("field", ["min_persona_pass_rate", "max_pass_rate_drop"])
+    @pytest.mark.parametrize("value", [-0.1, 1.5, float("nan")])
+    def test_rejects_values_outside_unit_interval(self, field: str, value: float) -> None:
+        with pytest.raises(ValueError, match=field):
+            ReadinessPolicy(**{field: value})
+
+    def test_defaults_add_no_thresholds(self) -> None:
+        policy = ReadinessPolicy()
+
+        assert policy.min_persona_pass_rate is None
+        assert policy.max_pass_rate_drop is None
+
+
+class TestVerdict:
+    def test_all_passed_is_go(self) -> None:
+        readiness = evaluate_readiness(_report(_scenario("t::a"), _scenario("t::b")))
+
+        assert readiness.verdict == "go"
+        assert readiness.blocking == []
+        assert readiness.baseline is None
+
+    @pytest.mark.parametrize(
+        ("scenarios", "reason"),
+        [
+            ((), "No scenarios executed"),
+            ((_scenario("t::a", "skipped"),), "No scenarios executed"),
+            ((_scenario("t::a", "failed"),), "1 scenario(s) failed or errored"),
+            (
+                (_scenario("t::a", "error"), _scenario("t::b", "failed")),
+                "2 scenario(s) failed or errored",
+            ),
+            ((_scenario("t::a", security_failure=True),), "1 security failure(s)"),
+            ((_scenario("t::a", duplicate_write_count=2),), "2 duplicate write(s)"),
+        ],
+    )
+    def test_blocking_reasons(self, scenarios: tuple[ScenarioReport, ...], reason: str) -> None:
+        readiness = evaluate_readiness(_report(*scenarios))
+
+        assert readiness.verdict == "no_go"
+        assert reason in readiness.blocking
+
+    def test_extra_notes_never_block(self) -> None:
+        readiness = evaluate_readiness(_report(_scenario("t::a")), notes=["heads up"])
+
+        assert readiness.verdict == "go"
+        assert readiness.notes == ["heads up"]
+
+
+class TestPersonaSegments:
+    def test_segments_are_weakest_first_and_skip_unexecuted(self) -> None:
+        report = _report(
+            _scenario("t::a", persona="novice", pass_rate=0.6),
+            _scenario("t::b", persona="novice"),
+            _scenario("t::c", persona="expert"),
+            _scenario("t::d", "skipped", persona="expert"),
+            _scenario("t::e", "failed", persona="frustrated"),
+            _scenario("t::f"),
+        )
+
+        segments = evaluate_readiness(report).persona_segments
+
+        assert [(s.persona, s.scenarios, s.passed) for s in segments] == [
+            ("frustrated", 1, 0),
+            ("novice", 2, 2),
+            ("expert", 1, 1),
+        ]
+        assert segments[1].pass_rate == pytest.approx(0.8)
+
+    @pytest.mark.parametrize(("threshold", "verdict"), [(0.9, "no_go"), (0.8, "go")])
+    def test_min_persona_pass_rate(self, threshold: float, verdict: str) -> None:
+        report = _report(
+            _scenario("t::a", persona="novice", pass_rate=0.8),
+            _scenario("t::b", persona="expert", pass_rate=1.0),
+        )
+
+        readiness = evaluate_readiness(
+            report,
+            policy=ReadinessPolicy(min_persona_pass_rate=threshold),
+        )
+
+        assert readiness.verdict == verdict
+        if verdict == "no_go":
+            assert readiness.blocking == ["Persona 'novice' pass rate 80% is below 90%"]
+
+
+class TestBaseline:
+    def test_regression_blocks_and_is_listed(self) -> None:
+        baseline = _report(_scenario("t::a"), _scenario("t::b"), run_id="run_before")
+        current = _report(_scenario("t::a", "failed"), _scenario("t::b"))
+
+        readiness = evaluate_readiness(current, baseline=baseline)
+
+        assert readiness.verdict == "no_go"
+        assert readiness.baseline is not None
+        assert readiness.baseline.baseline_run_id == "run_before"
+        assert readiness.baseline.regressions == ["t::a"]
+        assert "1 scenario(s) regressed since baseline run_before" in readiness.blocking
+
+    def test_already_failing_scenario_is_not_a_regression(self) -> None:
+        baseline = _report(_scenario("t::a", "failed"))
+        current = _report(_scenario("t::a", "failed"))
+
+        readiness = evaluate_readiness(current, baseline=baseline)
+
+        assert readiness.baseline is not None
+        assert readiness.baseline.regressions == []
+
+    def test_pass_rate_drop_is_a_note_by_default(self) -> None:
+        baseline = _report(_scenario("t::a", pass_rate=1.0), run_id="before")
+        current = _report(_scenario("t::a", pass_rate=0.8))
+
+        readiness = evaluate_readiness(current, baseline=baseline)
+
+        assert readiness.verdict == "go"
+        assert readiness.notes == ["Pass rate dropped: t::a: 100% -> 80%"]
+
+    @pytest.mark.parametrize(("max_drop", "verdict"), [(0.1, "no_go"), (0.3, "go")])
+    def test_max_pass_rate_drop_blocks(self, max_drop: float, verdict: str) -> None:
+        baseline = _report(_scenario("t::a", pass_rate=1.0))
+        current = _report(_scenario("t::a", pass_rate=0.8))
+
+        readiness = evaluate_readiness(
+            current,
+            baseline=baseline,
+            policy=ReadinessPolicy(max_pass_rate_drop=max_drop),
+        )
+
+        assert readiness.verdict == verdict
+
+    def test_fingerprint_changes_and_missing_scenarios_are_notes(self) -> None:
+        baseline = _report(
+            _scenario("t::a"),
+            _scenario("t::gone"),
+            _scenario("t::skipped", "skipped"),
+            model="model-a",
+            knowledge_version="kb-1",
+        )
+        current = _report(_scenario("t::a"), model="model-b", knowledge_version="kb-2")
+
+        readiness = evaluate_readiness(current, baseline=baseline)
+
+        assert readiness.verdict == "go"
+        assert readiness.baseline is not None
+        assert readiness.baseline.missing_scenarios == ["t::gone"]
+        assert readiness.notes == [
+            "model changed: model-a -> model-b",
+            "knowledge_version changed: kb-1 -> kb-2",
+            "1 baseline scenario(s) were not run",
+        ]
+
+    @pytest.mark.parametrize(
+        ("before", "after", "note"),
+        [
+            (100, 150, "Token usage increased 50% (100 -> 150)"),
+            (100, 110, None),
+            (None, 150, None),
+        ],
+    )
+    def test_token_usage_note(self, before: int | None, after: int, note: str | None) -> None:
+        baseline = _report(_scenario("t::a", token_usage=before))
+        current = _report(_scenario("t::a", token_usage=after))
+
+        readiness = evaluate_readiness(current, baseline=baseline)
+
+        assert readiness.notes == ([note] if note else [])
+
+
+class TestReportRoundTrip:
+    def test_json_writer_records_regression_fields_and_knowledge_version(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        output = tmp_path / "report.json"
+        writer = JsonReportWriter(
+            config=AgentTestConfig(agent_id="demo", knowledge_version="kb-7"),
+            output_path=output,
+        )
+        writer.record_scenario(
+            name="test_refund",
+            nodeid="t::refund",
+            passed=True,
+            duration_ms=5,
+            markers=[],
+            regression_evaluation=_evaluation(pass_rate=0.8, runs=5),
+        )
+        writer.write()
+
+        loaded = load_report(output)
+        scenario = loaded.scenarios[0]
+
+        assert loaded.knowledge_version == "kb-7"
+        assert (scenario.case_id, scenario.goal, scenario.persona) == (
+            "refund_status",
+            "Customer learns refund status",
+            "frustrated_customer",
+        )
+        assert (scenario.pass_rate, scenario.min_pass_rate, scenario.run_count) == (0.8, 0.6, 5)
+
+    def test_html_shows_readiness_and_behavior_rows(self, tmp_path: Path) -> None:
+        baseline = _report(_scenario("t::a"), run_id="run_before", model="m1")
+        report = _report(
+            _scenario(
+                "t::a",
+                "failed",
+                goal="Refund <status>",
+                persona="novice",
+                pass_rate=0.4,
+                min_pass_rate=0.8,
+                run_count=5,
+            ),
+            model="m2",
+            knowledge_version="kb-9",
+        )
+        report.readiness = evaluate_readiness(report, baseline=baseline)
+
+        html = HtmlReportWriter(tmp_path / "report.html").write(report).read_text("utf-8")
+
+        assert 'class="card readiness readiness-no-go"' in html
+        assert "NO-GO" in html
+        assert "1 scenario(s) regressed since baseline run_before" in html
+        assert "model changed: m1 -&gt; m2" in html
+        assert "Knowledge version" in html
+        assert "kb-9" in html
+        assert "Refund &lt;status&gt;" in html
+        assert "40% (required 80%)" in html
+
+    @pytest.mark.parametrize(
+        ("readiness", "expected", "absent"),
+        [
+            ("go", 'class="card readiness readiness-go"', "NO-GO"),
+            (None, "Agent test report", "Release readiness"),
+        ],
+    )
+    def test_html_readiness_optional(
+        self,
+        tmp_path: Path,
+        readiness: str | None,
+        expected: str,
+        absent: str,
+    ) -> None:
+        report = _report(_scenario("t::a"))
+        if readiness:
+            report.readiness = evaluate_readiness(report)
+
+        html = HtmlReportWriter(tmp_path / "r.html").write(report).read_text("utf-8")
+
+        assert expected in html
+        assert absent not in html
+
+
+def _session_state(tmp_path: Path, **fields: Any) -> tuple[MagicMock, pytest_plugin.PluginState]:
+    writer = JsonReportWriter(config=AgentTestConfig(), output_path=tmp_path / "report.json")
+    writer.record_scenario(name="a", nodeid="t::a", passed=True, duration_ms=1, markers=[])
+    state = pytest_plugin.PluginState(agent_config=AgentTestConfig(), json_writer=writer, **fields)
+    session = MagicMock()
+    session.config.stash = pytest.Stash()
+    session.config.stash[pytest_plugin._STASH_KEY] = state
+    session.exitstatus = pytest.ExitCode.OK
+    return session, state
+
+
+class TestPluginSession:
+    @pytest.mark.parametrize(
+        ("baseline_content", "note_prefix"),
+        [(None, "Baseline report not found"), ("{not json", "Baseline report unreadable")],
+    )
+    def test_bad_baseline_becomes_a_note(
+        self,
+        tmp_path: Path,
+        baseline_content: str | None,
+        note_prefix: str,
+    ) -> None:
+        baseline = tmp_path / "baseline.json"
+        if baseline_content is not None:
+            baseline.write_text(baseline_content, encoding="utf-8")
+        session, _ = _session_state(tmp_path, baseline_path=baseline)
+
+        pytest_plugin.pytest_sessionfinish(session, pytest.ExitCode.OK)
+
+        payload = json.loads((tmp_path / "report.json").read_text(encoding="utf-8"))
+        assert payload["readiness"]["verdict"] == "go"
+        assert payload["readiness"]["notes"][0].startswith(note_prefix)
+        assert session.exitstatus == pytest.ExitCode.OK
+
+    def test_valid_baseline_is_compared(self, tmp_path: Path) -> None:
+        baseline = tmp_path / "baseline.json"
+        baseline.write_text(_report(_scenario("t::a"), run_id="run_b").model_dump_json())
+        session, _ = _session_state(tmp_path, baseline_path=baseline)
+
+        pytest_plugin.pytest_sessionfinish(session, pytest.ExitCode.OK)
+
+        payload = json.loads((tmp_path / "report.json").read_text(encoding="utf-8"))
+        assert payload["readiness"]["baseline"]["baseline_run_id"] == "run_b"
+
+    @pytest.mark.parametrize(
+        ("enforce", "exitstatus", "expected"),
+        [
+            (True, pytest.ExitCode.OK, pytest.ExitCode.TESTS_FAILED),
+            (False, pytest.ExitCode.OK, pytest.ExitCode.OK),
+            (True, pytest.ExitCode.INTERRUPTED, pytest.ExitCode.INTERRUPTED),
+        ],
+    )
+    def test_enforcement_is_opt_in(
+        self,
+        tmp_path: Path,
+        enforce: bool,
+        exitstatus: pytest.ExitCode,
+        expected: pytest.ExitCode,
+    ) -> None:
+        session, _ = _session_state(
+            tmp_path,
+            readiness_policy=ReadinessPolicy(min_persona_pass_rate=1.0),
+            enforce_readiness=enforce,
+        )
+        session.exitstatus = exitstatus
+        state = session.config.stash[pytest_plugin._STASH_KEY]
+        assert state.json_writer is not None
+        state.json_writer.scenarios[0].persona = "novice"
+        state.json_writer.scenarios[0].pass_rate = 0.5
+
+        pytest_plugin.pytest_sessionfinish(session, exitstatus)
+
+        assert session.exitstatus == expected
+
+    def test_attach_regression_evaluation_ignores_missing_state(self) -> None:
+        config = MagicMock()
+        config.stash = pytest.Stash()
+
+        pytest_plugin.ScenarioResultRecorder(config, "t::a").attach_regression_evaluation(
+            _evaluation()
+        )
+
+        assert pytest_plugin._STASH_KEY not in config.stash
+
+
+def test_cli_enforces_persona_readiness_end_to_end(
+    pytester: pytest.Pytester,
+    tmp_path: Path,
+) -> None:
+    json_path = tmp_path / "readiness.json"
+    pytester.makepyfile(
+        """
+from agent_test_kit.regression import RegressionEvaluation, RegressionRunEvaluation
+
+
+def test_refund(agent_scenario):
+    agent_scenario.attach_regression_evaluation(
+        RegressionRunEvaluation(
+            case_id="refund",
+            goal="Refund status",
+            persona="novice",
+            passed=True,
+            pass_rate=0.6,
+            min_pass_rate=0.6,
+            runs=(RegressionEvaluation(case_id="refund", passed=True),),
+        )
+    )
+"""
+    )
+
+    result = pytester.runpytest(
+        f"--agent-report-json={json_path}",
+        "--agent-min-persona-pass-rate=0.8",
+        "--agent-enforce-readiness",
+        "-q",
+        "-p",
+        "no:cacheprovider",
+    )
+
+    result.assert_outcomes(passed=1)
+    assert result.ret == pytest.ExitCode.TESTS_FAILED
+    payload = json.loads(json_path.read_text(encoding="utf-8"))
+    assert payload["readiness"]["verdict"] == "no_go"
+    assert payload["readiness"]["persona_segments"][0]["persona"] == "novice"
+    assert payload["scenarios"][0]["pass_rate"] == 0.6

From da740831626bd22ded758e1631a4fbf4f527fd07 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Thu, 24 Sep 2026 18:08:43 +0000
Subject: [PATCH 6/6] Document release readiness options in README

Co-authored-by: Dor Ingber  <Doringber@users.noreply.github.com>
---
 README.md | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/README.md b/README.md
index 49e7d70..ebd1263 100644
--- a/README.md
+++ b/README.md
@@ -636,6 +636,25 @@ pytest tests/ \
 
 Tool-flow data appears when you attach execution results — via fixtures, `agent_scenario.attach_execution_result(result)`, or `store_execution_result(...)`.
 
+**Release readiness.** Every report opens with a **GO / NO-GO** block: evidence for a human release decision, not a replacement for it. Failed or errored scenarios, security failures, and duplicate writes block. Everything else is a note unless you set a threshold.
+
+| Option | Effect |
+|--------|--------|
+| `--agent-baseline=reports/main.json` | Compare with a previous JSON report: regressions block; pass-rate drops, scenarios not run, token growth above 20%, and fingerprint changes (`model`, `prompt_version`, `agent_version`, `knowledge_version`) are notes. A missing baseline is a note. |
+| `--agent-min-persona-pass-rate=0.8` | Block when any persona's pass rate is below the value |
+| `--agent-enforce-readiness` | Fail the pytest session on NO-GO (default: report only) |
+
+Attach goal-based evaluations so the report can segment by persona:
+
+```python
+async def test_refund_case(agent_client, agent_scenario):
+    evaluation = await case.run(agent_client)
+    agent_scenario.attach_regression_evaluation(evaluation)
+    evaluation.assert_passed()
+```
+
+Set `AGENT_TEST_KNOWLEDGE_VERSION` (RAG index or knowledge-base version) next to model and prompt version so answer drift can be traced to its source. To evaluate outside pytest: `evaluate_readiness(load_report(path), baseline=load_report(base), policy=ReadinessPolicy(max_pass_rate_drop=0.1))`.
+
 ### Step 12 — Wire into CI
 
 ```bash
@@ -711,6 +730,7 @@ from agent_test_kit import (
     JsonReportWriter,
     OptionalStep,
     PromptReviewClient,
+    ReadinessPolicy,
     RegressionRunEvaluation,
     RepeatedExecutionResult,
     RequiredFieldsScorer,
@@ -718,8 +738,10 @@ from agent_test_kit import (
     Score,
     Scorer,
     ToolStep,
+    evaluate_readiness,
     get_profile,
     load_regression_cases,
+    load_report,
     response_text,
     run_repeatedly,
     run_verifiers,