Files
cleveragents-core/benchmarks/autonomy_guardrails_bench.py
CoreRasurae 53455275ba
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 18s
CI / lint (pull_request) Successful in 23s
CI / security (pull_request) Successful in 31s
CI / typecheck (pull_request) Successful in 35s
CI / integration_tests (pull_request) Successful in 3m46s
CI / unit_tests (pull_request) Successful in 11m44s
CI / docker (pull_request) Successful in 38s
CI / benchmark-regression (pull_request) Successful in 27m39s
CI / coverage (pull_request) Successful in 55m41s
CI / build (push) Successful in 15s
CI / quality (push) Successful in 16s
CI / lint (push) Successful in 19s
CI / typecheck (push) Successful in 32s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 41s
CI / integration_tests (push) Successful in 2m55s
CI / unit_tests (push) Successful in 11m43s
CI / docker (push) Successful in 40s
CI / benchmark-publish (push) Successful in 16m11s
CI / coverage (push) Successful in 44m49s
feat(automation): add autonomy guardrails and audit trail
Add runtime autonomy constraints (max steps, tool budget, required
confirmations) and a structured audit trail for plan execution.

New domain models:
- AutonomyGuardrails: enforces step limits, tool budgets, and
  confirmation gates with validators and check methods
- GuardrailAuditEntry: records each enforcement event with timestamp,
  event type, guard name, result, reason, and context
- GuardrailAuditTrail: ordered collection of audit entries persisted
  to plan metadata

New service:
- AutonomyGuardrailService: high-level service for configuring
  guardrails per plan, checking constraints, recording audit entries,
  and serializing/restoring state via plan metadata

Tests:
- Behave: 69 scenarios covering model validation, step/budget/
  confirmation checks, audit trail recording, and service operations
- Robot: 8 test cases for autonomy guardrail CLI flag smoke testing
- ASV: 6 benchmark suites measuring enforcement overhead

Documentation:
- Updated docs/reference/automation_profiles.md with guardrail fields,
  enforcement behavior, audit trail schema, and event type reference

ISSUES CLOSED: #204
2026-02-28 22:06:57 +00:00

286 lines
9.4 KiB
Python

"""ASV benchmarks for autonomy guardrails enforcement overhead.
Measures the performance of:
- AutonomyGuardrails model construction
- Step limit checks
- Tool budget checks
- Confirmation checks
- Wall-clock time limit checks
- Actor tool-call limit checks
- Audit trail entry creation, appending, and eviction
- AutonomyGuardrailService operations
- Metadata serialization and deserialization
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.application.services.autonomy_guardrail_service import (
AutonomyGuardrailService,
)
from cleveragents.domain.models.core.autonomy_guardrails import (
ActorLimits,
AutonomyGuardrails,
GuardrailAuditEntry,
GuardrailAuditTrail,
GuardrailEventType,
GuardrailResult,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.autonomy_guardrail_service import (
AutonomyGuardrailService,
)
from cleveragents.domain.models.core.autonomy_guardrails import (
ActorLimits,
AutonomyGuardrails,
GuardrailAuditEntry,
GuardrailAuditTrail,
GuardrailEventType,
GuardrailResult,
)
class GuardrailsConstructionSuite:
"""Benchmark AutonomyGuardrails model construction."""
def time_guardrails_construction_full(self) -> None:
"""Benchmark fully-populated guardrails creation."""
AutonomyGuardrails(
max_steps=100,
tool_budget=500.0,
required_confirmations=["deploy", "delete", "write"],
)
def time_guardrails_construction_minimal(self) -> None:
"""Benchmark minimal guardrails creation."""
AutonomyGuardrails()
def time_guardrails_construction_steps_only(self) -> None:
"""Benchmark guardrails with only step limit."""
AutonomyGuardrails(max_steps=50)
def time_guardrails_construction_budget_only(self) -> None:
"""Benchmark guardrails with only budget."""
AutonomyGuardrails(tool_budget=100.0)
class StepLimitCheckSuite:
"""Benchmark step limit enforcement checks."""
def setup(self) -> None:
"""Create guardrails for benchmarking."""
self.limited = AutonomyGuardrails(max_steps=100)
self.unlimited = AutonomyGuardrails()
def time_step_limit_allowed(self) -> None:
"""Benchmark step check when allowed."""
self.limited.step_count = 50
self.limited.check_step_limit()
def time_step_limit_denied(self) -> None:
"""Benchmark step check when denied."""
self.limited.step_count = 100
self.limited.check_step_limit()
def time_step_limit_unlimited(self) -> None:
"""Benchmark step check with no limit."""
self.unlimited.step_count = 10000
self.unlimited.check_step_limit()
class BudgetCheckSuite:
"""Benchmark tool budget enforcement checks."""
def setup(self) -> None:
"""Create guardrails for benchmarking."""
self.budgeted = AutonomyGuardrails(tool_budget=100.0)
self.unlimited = AutonomyGuardrails()
def time_budget_allowed(self) -> None:
"""Benchmark budget check when allowed."""
self.budgeted.budget_spent = 50.0
self.budgeted.check_tool_budget(10.0)
def time_budget_denied(self) -> None:
"""Benchmark budget check when denied."""
self.budgeted.budget_spent = 95.0
self.budgeted.check_tool_budget(10.0)
def time_budget_unlimited(self) -> None:
"""Benchmark budget check with no limit."""
self.unlimited.check_tool_budget(999.0)
class ConfirmationCheckSuite:
"""Benchmark confirmation requirement checks."""
def setup(self) -> None:
"""Create guardrails for benchmarking."""
self.with_confirmations = AutonomyGuardrails(
required_confirmations=[
"deploy",
"delete",
"write",
"apply",
"revert",
],
)
self.no_confirmations = AutonomyGuardrails()
def time_confirmation_required(self) -> None:
"""Benchmark check when confirmation is required."""
self.with_confirmations.check_confirmation_required("deploy")
def time_confirmation_not_required(self) -> None:
"""Benchmark check when confirmation is not required."""
self.with_confirmations.check_confirmation_required("read")
def time_confirmation_empty_list(self) -> None:
"""Benchmark check with empty confirmation list."""
self.no_confirmations.check_confirmation_required("deploy")
class AuditTrailSuite:
"""Benchmark audit trail operations."""
def setup(self) -> None:
"""Create audit trail for benchmarking."""
self.trail = GuardrailAuditTrail()
self.entry = GuardrailAuditEntry(
event_type=GuardrailEventType.STEP_ALLOWED,
guard_name="step_limit",
result=GuardrailResult.ALLOWED,
)
def time_audit_entry_creation(self) -> None:
"""Benchmark audit entry creation."""
GuardrailAuditEntry(
event_type=GuardrailEventType.STEP_ALLOWED,
guard_name="step_limit",
result=GuardrailResult.ALLOWED,
)
def time_audit_trail_append(self) -> None:
"""Benchmark appending to audit trail."""
self.trail.add_entry(self.entry)
def time_audit_trail_counts(self) -> None:
"""Benchmark counting allowed/denied entries."""
_ = self.trail.allowed_count
_ = self.trail.denied_count
class ServiceSuite:
"""Benchmark AutonomyGuardrailService operations."""
def setup(self) -> None:
"""Create service for benchmarking."""
self.service = AutonomyGuardrailService()
self.guardrails = AutonomyGuardrails(
max_steps=100,
tool_budget=500.0,
required_confirmations=["deploy"],
)
self.service.configure_guardrails("bench-plan", self.guardrails)
def time_service_check_step(self) -> None:
"""Benchmark service step limit check."""
self.service.check_step_limit("bench-plan", 50)
def time_service_check_budget(self) -> None:
"""Benchmark service budget check."""
self.service.check_tool_budget("bench-plan", 1.0)
def time_service_check_confirmation(self) -> None:
"""Benchmark service confirmation check."""
self.service.check_confirmation_required("bench-plan", "deploy")
def time_service_get_metadata(self) -> None:
"""Benchmark service metadata serialization."""
self.service.get_audit_trail_as_metadata("bench-plan")
def time_service_configure(self) -> None:
"""Benchmark service guardrail configuration."""
self.service.configure_guardrails(
"new-plan",
AutonomyGuardrails(max_steps=50),
)
def time_service_check_wall_clock(self) -> None:
"""Benchmark service wall-clock check."""
self.service.check_wall_clock("bench-plan")
def time_service_check_actor_calls(self) -> None:
"""Benchmark service actor tool-call check."""
self.service.check_actor_tool_calls("bench-plan", 5)
class WallClockCheckSuite:
"""Benchmark wall-clock time limit checks."""
def setup(self) -> None:
"""Create guardrails for benchmarking."""
self.with_limit = AutonomyGuardrails(max_wall_clock_seconds=3600.0)
self.with_limit.mark_started()
self.no_limit = AutonomyGuardrails()
def time_wall_clock_allowed(self) -> None:
"""Benchmark wall-clock check when allowed."""
self.with_limit.check_wall_clock()
def time_wall_clock_no_limit(self) -> None:
"""Benchmark wall-clock check with no limit."""
self.no_limit.check_wall_clock()
class ActorLimitCheckSuite:
"""Benchmark per-actor tool-call limit checks."""
def setup(self) -> None:
"""Create guardrails for benchmarking."""
self.with_limit = AutonomyGuardrails(
actor_limits=ActorLimits(max_tool_calls_per_invocation=100),
)
self.no_limit = AutonomyGuardrails()
def time_actor_limit_allowed(self) -> None:
"""Benchmark actor limit check when allowed."""
self.with_limit.check_actor_tool_calls(50)
def time_actor_limit_denied(self) -> None:
"""Benchmark actor limit check when denied."""
self.with_limit.check_actor_tool_calls(100)
def time_actor_limit_unlimited(self) -> None:
"""Benchmark actor limit check with no limit."""
self.no_limit.check_actor_tool_calls(10000)
class AuditTrailEvictionSuite:
"""Benchmark audit trail eviction under load."""
def setup(self) -> None:
"""Create a full audit trail for benchmarking eviction."""
self.trail = GuardrailAuditTrail(max_entries=100)
for _ in range(100):
self.trail.add_entry(
GuardrailAuditEntry(
event_type=GuardrailEventType.STEP_ALLOWED,
guard_name="step_limit",
result=GuardrailResult.ALLOWED,
)
)
self.entry = GuardrailAuditEntry(
event_type=GuardrailEventType.STEP_ALLOWED,
guard_name="step_limit",
result=GuardrailResult.ALLOWED,
)
def time_eviction_append(self) -> None:
"""Benchmark appending to a full trail (triggers eviction)."""
self.trail.add_entry(self.entry)