Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions docs/AGENT_E2E_FULL_FLOW.md
Original file line number Diff line number Diff line change
Expand Up @@ -252,8 +252,31 @@ Example upload step (requires `aws-actions/configure-aws-credentials` + `s3:PutO
"s3://${REPORT_BUCKET}/${REPORT_PREFIX}/agent-test-kit-post-deploy.html"
aws s3 cp reports/agent-test-kit-post-deploy.html \
"s3://${REPORT_BUCKET}/agent-ci-reports/${{ github.repository }}/latest.html"
aws s3 cp reports/agent-test-kit-post-deploy.json \
"s3://${REPORT_BUCKET}/${REPORT_PREFIX}/agent-test-kit-post-deploy.json"
```

### Step 10 — Twice-daily monitor reads the same JSON

The production monitor does not re-run pytest. It reads the JSON report from
step 9 (`--agent-report-json`) and keeps the kit's GO / NO-GO verdict.

Two phases stay separate:

| Phase | Marker | What the monitor calls it |
|-------|--------|---------------------------|
| Build / unit | `agent_unit` | `unit` |
| Post-deploy E2E | `agent_e2e` | `post_deploy` |

`monitor_summary(report)` returns that slice. The keys are
`schema=agent-test-kit.monitor/v1`, `verdict` (`go` or `no_go`), `blocking`,
`phase`, `pass_rate`, `security_failures`, `eval_failures`, and
`guardrail_failures`. A failed scenario, a security failure, or a duplicate
write is `no_go`. Notes never flip the verdict.

Upload the JSON next to the HTML. The monitor's dry run can load that file
with no AWS account and no live agent.

---

## 4. Which client to use
Expand Down
2 changes: 2 additions & 0 deletions src/agent_test_kit/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,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.monitor_contract import monitor_summary
from agent_test_kit.reporting.readiness import ReadinessPolicy, evaluate_readiness, load_report
from agent_test_kit.scoring import (
ForbiddenTermsScorer,
Expand Down Expand Up @@ -149,6 +150,7 @@
"guardrail",
"load_regression_cases",
"load_report",
"monitor_summary",
"get_profile",
"must_open_pull_request",
"register_eval",
Expand Down
3 changes: 3 additions & 0 deletions src/agent_test_kit/reporting/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
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.monitor_contract import MonitorSummary, monitor_summary

__all__ = [
"HtmlReportWriter",
"JsonReportWriter",
"MarkdownStatusReportWriter",
"MonitorSummary",
"monitor_summary",
]
86 changes: 86 additions & 0 deletions src/agent_test_kit/reporting/monitor_contract.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
"""Stable slice of a kit report for the twice-daily agent monitor.

The monitor does not import this package. It reads the same keys from the
JSON written by ``--agent-report-json``, or from :func:`monitor_summary`.
Keep the keys stable.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
from typing import Any, Literal

from agent_test_kit.models.report import AgentTestRunReport
from agent_test_kit.reporting.readiness import evaluate_readiness

MONITOR_SCHEMA = "agent-test-kit.monitor/v1"
Phase = Literal["unit", "post_deploy"]


@dataclass(frozen=True, slots=True)
class MonitorSummary:
"""GO / NO-GO evidence the monitor renders next to production metrics."""

agent_id: str
run_id: str
environment: str
phase: Phase
verdict: Literal["go", "no_go"]
blocking: tuple[str, ...]
notes: tuple[str, ...]
scenarios: int
passed: int
failed: int
pass_rate: float | None
security_failures: int
eval_failures: tuple[str, ...]
guardrail_failures: tuple[str, ...]

def as_dict(self) -> dict[str, Any]:
"""Return the JSON object the monitor accepts."""
payload = asdict(self)
payload["schema"] = MONITOR_SCHEMA
return payload


def monitor_summary(report: AgentTestRunReport) -> MonitorSummary:
"""Summarize a kit report the way the twice-daily monitor will read it.

The verdict is the kit's release readiness. Failed evals and guardrails
are listed even when the scenario itself stayed green.
"""
readiness = report.readiness if report.readiness is not None else evaluate_readiness(report)
executed = [scenario for scenario in report.scenarios if scenario.status != "skipped"]
passed = sum(1 for scenario in executed if scenario.status == "passed")
failed = sum(1 for scenario in executed if scenario.status in {"failed", "error"})
phase: Phase = (
"post_deploy"
if any("agent_e2e" in scenario.markers for scenario in report.scenarios)
else "unit"
)
return MonitorSummary(
agent_id=report.agent_id,
run_id=report.run_id,
environment=report.environment,
phase=phase,
verdict=readiness.verdict,
blocking=tuple(readiness.blocking),
notes=tuple(readiness.notes),
scenarios=len(executed),
passed=passed,
failed=failed,
pass_rate=(passed / len(executed)) if executed else None,
security_failures=sum(1 for scenario in executed if scenario.security_failure),
eval_failures=tuple(
f"{scenario.nodeid}: {item.name}"
for scenario in executed
for item in scenario.evals
if not item.passed
),
guardrail_failures=tuple(
f"{scenario.nodeid}: {item.name}"
for scenario in executed
for item in scenario.guardrails
if not item.passed
),
)
99 changes: 99 additions & 0 deletions tests/test_monitor_contract.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
"""The monitor contract stays aligned with release readiness."""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any

import pytest

from agent_test_kit.models.report import (
AgentTestRunReport,
EvalCheckReport,
GuardrailCheckReport,
ScenarioReport,
ScenarioStatus,
)
from agent_test_kit.reporting.monitor_contract import MONITOR_SCHEMA, monitor_summary

pytestmark = pytest.mark.agent_unit

_NOW = datetime(2026, 11, 8, 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, **fields: Any) -> AgentTestRunReport:
report = AgentTestRunReport(
run_id="run_now",
agent_id="pr-review",
environment="ci",
started_at=_NOW,
completed_at=_NOW,
scenarios=list(scenarios),
**fields,
)
report.finalize_metrics()
return report


def test_unit_pass_is_go() -> None:
summary = monitor_summary(_report(_scenario("tests/test_review.py::test_mock")))

assert summary.verdict == "go"
assert summary.phase == "unit"
assert summary.pass_rate == 1.0
assert summary.as_dict()["schema"] == MONITOR_SCHEMA


def test_failed_scenario_is_no_go() -> None:
summary = monitor_summary(_report(_scenario("tests/test_review.py::test_mock", "failed")))

assert summary.verdict == "no_go"
assert summary.failed == 1
assert any("failed or errored" in reason for reason in summary.blocking)


def test_live_marker_selects_the_post_deploy_phase() -> None:
summary = monitor_summary(
_report(_scenario("tests/agent_e2e/test_live.py::test_health", markers=["agent_e2e"]))
)

assert summary.phase == "post_deploy"
assert summary.verdict == "go"


def test_eval_and_guardrail_failures_are_listed_without_changing_a_green_scenario() -> None:
summary = monitor_summary(
_report(
_scenario(
"tests/test_review.py::test_mock",
evals=[
EvalCheckReport(name="terms", passed=False, score=0.2, threshold=0.8),
],
guardrails=[GuardrailCheckReport(name="no-secrets", passed=False)],
)
)
)

assert summary.verdict == "go"
assert summary.eval_failures == ("tests/test_review.py::test_mock: terms",)
assert summary.guardrail_failures == ("tests/test_review.py::test_mock: no-secrets",)


def test_security_failure_blocks() -> None:
summary = monitor_summary(
_report(_scenario("tests/test_review.py::test_mock", security_failure=True))
)

assert summary.verdict == "no_go"
assert summary.security_failures == 1
Loading