diff --git a/benchmarks/autonomy_guardrails_bench.py b/benchmarks/autonomy_guardrails_bench.py new file mode 100644 index 000000000..b3e5d4bee --- /dev/null +++ b/benchmarks/autonomy_guardrails_bench.py @@ -0,0 +1,285 @@ +"""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) diff --git a/docs/reference/automation_profiles.md b/docs/reference/automation_profiles.md index 336b14c15..c1f285791 100644 --- a/docs/reference/automation_profiles.md +++ b/docs/reference/automation_profiles.md @@ -271,3 +271,123 @@ Legacy automation levels are mapped to built-in profiles: | `full_auto` | `full-auto` | Use `--automation-profile ` instead of `--automation-level`. + +## Autonomy Guardrails + +Autonomy guardrails extend the automation guard framework with runtime +constraints that enforce step limits, tool budgets, and required human +confirmations during plan execution. Unlike the phase-transition thresholds +which gate entire phases, autonomy guardrails operate at the individual +step and tool invocation level. + +### Guardrail Fields + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `max_steps` | `int \| None` | `None` | Maximum execution steps allowed. `None` means unlimited. | +| `tool_budget` | `float \| None` | `None` | Maximum cumulative cost for tool invocations. `None` means unlimited. A value of `0.0` blocks any positive cost. | +| `max_wall_clock_seconds` | `float \| None` | `None` | Maximum wall-clock time in seconds for plan execution. `None` means unlimited. | +| `actor_limits` | `ActorLimits` | `ActorLimits()` | Per-actor runtime limits (see below). | +| `required_confirmations` | `list[str]` | `[]` | Operation names requiring human confirmation before execution (case-insensitive). | +| `step_count` | `int` | `0` | Current execution step counter (managed by the service). | +| `budget_spent` | `float` | `0.0` | Current cumulative budget spent (managed by the service). | +| `start_time` | `str \| None` | `None` | ISO-8601 timestamp of when plan execution started (for wall-clock tracking). | + +### Actor Limits + +Per-actor constraints that limit tool calls within a single actor invocation: + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `max_tool_calls_per_invocation` | `int \| None` | `None` | Maximum tool calls allowed per actor invocation. `None` means unlimited. | +| `max_retries_per_failure` | `int \| None` | `None` | Maximum retries per tool failure within an actor step. `None` means unlimited. | + +### Enforcement Behavior + +Guardrails are enforced at multiple points during plan execution: + +- **Pre-flight** — Before execution begins, `PlanExecutor._guard_execute()` calls `_enforce_guardrails()` which checks wall-clock time and the current step limit. If either check fails, execution is blocked with a `PlanError` before any decisions are processed. +- **Per-decision** — Before each decision in the execution loop, `_enforce_guardrails_per_step()` increments the step counter, checks the step limit, and rechecks wall-clock time. If any check fails, execution halts with a `PlanError`. +- **Start-time tracking** — On the first execution, `_enforce_guardrails()` automatically calls `mark_started()` on the guardrails model to record the wall-clock start time (if not already set). + +The individual guardrail checks are: + +1. **Step limit** — The `step_count` is compared against `max_steps`. If equal or exceeded, the step is **blocked** and an audit entry is recorded. +2. **Tool budget** — The projected cost (`budget_spent + tool_cost`) is compared against `tool_budget`. If exceeded, the invocation is **blocked** and an audit entry is recorded. When allowed, the cost is recorded **before** the audit entry so the trail reflects the post-deduction budget state. +3. **Wall-clock time** — If `max_wall_clock_seconds` is set and `start_time` is recorded, the elapsed wall-clock time is checked. If the limit is exceeded, the step is **blocked**. Malformed `start_time` values are treated as a block (defense-in-depth) rather than raising an unhandled exception. +4. **Actor tool-call limit** — The actor's current tool call count is compared against `actor_limits.max_tool_calls_per_invocation`. If exceeded, the call is **blocked**. +5. **Retry-per-failure limit** — The current retry count is compared against `actor_limits.max_retries_per_failure`. If equal or exceeded, the retry is **blocked** and an audit entry is recorded. When `max_retries_per_failure` is `None`, retries are unlimited. +6. **Required confirmations** — The operation name is checked (case-insensitively) against `required_confirmations`. If matched, the operation is **blocked** pending human confirmation. + +If no guardrail blocks the invocation, it proceeds and the step counter and budget are updated. + +### Input Validation + +- `start_time` is validated on assignment to ensure it is a parseable ISO-8601 timestamp. Malformed strings are rejected with a `ValidationError`. +- `max_retries_per_failure` must be non-negative (`>= 0`) if set; `None` means unlimited. +- `max_steps` must be positive (`>= 1`) if set. +- `tool_budget` must be non-negative (`>= 0.0`) if set. + +### Audit Trail + +Every guardrail enforcement event is recorded in a `GuardrailAuditTrail` which is persisted to plan metadata. The audit trail provides a bounded, ordered record of autonomy-related decisions. When the number of entries exceeds `max_entries` (default: 10,000), the oldest entries are evicted to prevent unbounded memory growth. + +#### Audit Entry Fields + +| Field | Type | Description | +|-------|------|-------------| +| `timestamp` | `str` | ISO-8601 timestamp of the enforcement event. | +| `event_type` | `GuardrailEventType` | Type of guardrail event (see below). | +| `guard_name` | `str` | Name of the guard that was evaluated. | +| `result` | `GuardrailResult` | Outcome: `allowed` or `denied`. | +| `reason` | `str \| None` | Human-readable reason for the decision. | +| `context` | `dict` | Additional metadata about the enforcement event. | + +#### Event Types + +| Event Type | Description | +|------------|-------------| +| `step_blocked` | A step was blocked due to the step limit. | +| `budget_blocked` | A tool invocation was blocked due to budget. | +| `confirmation_required` | An operation was blocked pending confirmation. | +| `step_allowed` | A step was allowed within the limit. | +| `budget_allowed` | A tool invocation was allowed within budget. | +| `confirmation_granted` | An operation was allowed (no confirmation needed). | +| `time_blocked` | A step was blocked due to the wall-clock time limit. | +| `time_allowed` | A step was allowed within the wall-clock time limit. | +| `actor_limit_blocked` | An actor tool call was blocked due to per-invocation limits. | +| `actor_limit_allowed` | An actor tool call was allowed within per-invocation limits. | +| `retry_blocked` | A retry was blocked due to the per-failure retry limit. | +| `retry_allowed` | A retry was allowed within the per-failure retry limit. | + +### Plan Metadata Persistence + +The guardrail state and audit trail are serialized to plan metadata using two keys: + +- `autonomy_guardrails` — The current guardrail configuration and counters. +- `guardrail_audit_trail` — The ordered list of audit entries. + +This allows guardrail state to survive across plan restarts and provides +a complete audit record for post-mortem analysis. + +### Example Configuration + +```yaml +name: acme/guarded-auto +description: Auto profile with autonomy guardrails +schema_version: "1.0" +auto_strategize: 0.0 +auto_execute: 0.0 +auto_apply: 1.0 + +guards: + max_tool_calls_per_step: 10 + max_total_cost: 100.0 + +# Autonomy guardrails (configured via service or CLI) +# max_steps: 50 +# tool_budget: 200.0 +# required_confirmations: +# - deploy +# - delete +``` diff --git a/features/autonomy_guardrails.feature b/features/autonomy_guardrails.feature new file mode 100644 index 000000000..77eb2d764 --- /dev/null +++ b/features/autonomy_guardrails.feature @@ -0,0 +1,401 @@ +Feature: Autonomy Guardrails and Audit Trail + As a plan executor + I want autonomy guardrails to enforce step limits, tool budgets, and confirmations + So that plan execution stays within defined safety boundaries with a full audit trail + + # ---- AutonomyGuardrails model validation ---- + + Scenario: Create guardrails with all fields + When I create autonomy guardrails with max_steps 10 and tool_budget 50.0 + Then the guardrails should have max_steps 10 + And the guardrails should have tool_budget 50.0 + And the guardrails step_count should be 0 + And the guardrails budget_spent should be 0.0 + + Scenario: Create guardrails with defaults + When I create autonomy guardrails with defaults + Then the guardrails should have max_steps None + And the guardrails should have tool_budget None + And the guardrails should have empty required_confirmations + + Scenario: Guardrails reject negative max_steps + When I try to create guardrails with max_steps 0 + Then a guardrails validation error should be raised + And the guardrails error should mention "max_steps" + + Scenario: Guardrails reject negative tool_budget + When I try to create guardrails with tool_budget -1.0 + Then a guardrails validation error should be raised + And the guardrails error should mention "tool_budget" + + # ---- Step limit checks ---- + + Scenario: Step limit allows within bounds + Given guardrails with max_steps 5 + When I check step limit at step 3 + Then the step check should be allowed + + Scenario: Step limit blocks at max + Given guardrails with max_steps 5 + When I check step limit at step 5 + Then the step check should be denied + + Scenario: Step limit blocks above max + Given guardrails with max_steps 3 + When I check step limit at step 10 + Then the step check should be denied + + Scenario: Unlimited steps always allowed + Given guardrails with no step limit + When I check step limit at step 1000 + Then the step check should be allowed + + # ---- Tool budget checks ---- + + Scenario: Budget allows within limit + Given guardrails with tool_budget 100.0 + When I check tool budget for cost 50.0 + Then the budget check should be allowed + + Scenario: Budget blocks when exceeded + Given guardrails with tool_budget 10.0 and budget_spent 8.0 + When I check tool budget for cost 5.0 + Then the budget check should be denied + + Scenario: Budget allows exact remaining + Given guardrails with tool_budget 10.0 and budget_spent 5.0 + When I check tool budget for cost 5.0 + Then the budget check should be allowed + + Scenario: Unlimited budget always allowed + Given guardrails with no budget limit + When I check tool budget for cost 999.0 + Then the budget check should be allowed + + Scenario: Budget rejects negative tool cost + Given guardrails with tool_budget 100.0 + When I try to check tool budget for cost -1.0 + Then a guardrails validation error should be raised + + # ---- Confirmation checks ---- + + Scenario: Confirmation required for listed operation + Given guardrails with required confirmations "deploy,delete" + When I check confirmation for operation "deploy" + Then confirmation should be required + + Scenario: Confirmation not required for unlisted operation + Given guardrails with required confirmations "deploy,delete" + When I check confirmation for operation "read" + Then confirmation should not be required + + Scenario: No confirmations required when list empty + Given guardrails with no required confirmations + When I check confirmation for operation "deploy" + Then confirmation should not be required + + # ---- Step counter and budget tracking ---- + + Scenario: Increment step counter + Given guardrails with max_steps 10 + When I increment the step counter 3 times + Then the step_count should be 3 + + Scenario: Record tool cost + Given guardrails with tool_budget 100.0 + When I record a tool cost of 25.0 + And I record a tool cost of 15.0 + Then the budget_spent should be 40.0 + + Scenario: Record cost rejects negative cost + Given guardrails with tool_budget 100.0 + When I try to record a negative cost + Then a guardrails validation error should be raised + + # ---- Audit trail ---- + + Scenario: Audit trail records entries + Given an empty audit trail + When I add a step_allowed entry to the audit trail + And I add a budget_blocked entry to the audit trail + Then the audit trail should have 2 entries + And the audit trail allowed_count should be 1 + And the audit trail denied_count should be 1 + + Scenario: Audit entry has correct fields + When I create an audit entry with event_type step_blocked and guard_name step_limit + Then the audit entry event_type should be "step_blocked" + And the audit entry guard_name should be "step_limit" + And the audit entry result should be "denied" + And the audit entry should have a timestamp + + # ---- AutonomyGuardrailService ---- + + Scenario: Service configures and retrieves guardrails + Given an autonomy guardrail service + When I configure guardrails for plan "plan-001" with max_steps 5 + Then the service should return guardrails for plan "plan-001" + And the service guardrails should have max_steps 5 + + Scenario: Service returns None for unconfigured plan + Given an autonomy guardrail service + Then the service should return None for plan "nonexistent" + + Scenario: Service step limit check records audit trail + Given an autonomy guardrail service with guardrails for plan "plan-002" + When I check step limit via service for plan "plan-002" at step 3 + Then the audit trail for plan "plan-002" should have 1 entry + + Scenario: Service budget check records cost and audit trail + Given an autonomy guardrail service with budget guardrails for plan "plan-003" + When I check tool budget via service for plan "plan-003" with cost 5.0 + Then the audit trail for plan "plan-003" should have 1 entry + And the service guardrails budget_spent should be 5.0 + + Scenario: Service confirmation check records audit trail + Given an autonomy guardrail service with confirmation guardrails for plan "plan-004" + When I check confirmation via service for plan "plan-004" operation "deploy" + Then the audit trail for plan "plan-004" should have 1 entry + + Scenario: Service serializes audit trail to metadata + Given an autonomy guardrail service with guardrails for plan "plan-005" + When I check step limit via service for plan "plan-005" at step 1 + Then the metadata for plan "plan-005" should contain the audit trail + + Scenario: Service loads guardrails from metadata + Given an autonomy guardrail service + When I load guardrails from metadata for plan "plan-006" + Then the service should return guardrails for plan "plan-006" + + Scenario: Service removes plan guardrails + Given an autonomy guardrail service with guardrails for plan "plan-007" + When I remove plan "plan-007" from the service + Then the service should return None for plan "plan-007" + + Scenario: Service check_step_limit returns True for unconfigured plan + Given an autonomy guardrail service + When I check step limit via service for plan "unknown" at step 100 + Then the step limit check should return True + + Scenario: Service check_tool_budget returns True for unconfigured plan + Given an autonomy guardrail service + When I check tool budget via service for plan "unknown" with cost 999.0 + Then the tool budget check should return True + + Scenario: Service check_confirmation returns False for unconfigured plan + Given an autonomy guardrail service + When I check confirmation via service for plan "unknown" operation "deploy" + Then the confirmation check should return False + + Scenario: Service configure_guardrails rejects empty plan_id + Given an autonomy guardrail service + When I try to configure guardrails with empty plan_id + Then a guardrails validation error should be raised + + Scenario: GuardrailEventType enumerates all event types + Then GuardrailEventType should have at least 10 members + + Scenario: GuardrailResult enumerates allowed and denied + Then GuardrailResult should have exactly 2 members + + # ---- Wall-clock time checks ---- + + Scenario: Wall-clock allows when no limit set + Given guardrails with no wall-clock limit + When I check wall-clock time + Then the wall-clock check should be allowed + + Scenario: Wall-clock allows when within limit + Given guardrails with wall-clock limit 60.0 seconds started recently + When I check wall-clock time + Then the wall-clock check should be allowed + + Scenario: Wall-clock blocks when expired + Given guardrails with wall-clock limit that has already expired + When I check wall-clock time + Then the wall-clock check should be denied + + Scenario: Service wall-clock check records audit trail + Given an autonomy guardrail service with wall-clock guardrails for plan "plan-wc" + When I check wall-clock via service for plan "plan-wc" + Then the audit trail for plan "plan-wc" should have 1 entry + + # ---- Actor tool-call limits ---- + + Scenario: Actor tool-call limit allows within bounds + Given guardrails with actor tool-call limit 5 + When I check actor tool calls with 3 current calls + Then the actor tool-call check should be allowed + + Scenario: Actor tool-call limit blocks at max + Given guardrails with actor tool-call limit 5 + When I check actor tool calls with 5 current calls + Then the actor tool-call check should be denied + + Scenario: Actor tool-call limit unlimited + Given guardrails with no actor tool-call limit + When I check actor tool calls with 1000 current calls + Then the actor tool-call check should be allowed + + Scenario: Service actor tool-call check records audit trail + Given an autonomy guardrail service with actor-limit guardrails for plan "plan-al" + When I check actor tool calls via service for plan "plan-al" with 3 calls + Then the audit trail for plan "plan-al" should have 1 entry + + # ---- Bounded audit trail eviction ---- + + Scenario: Audit trail evicts oldest entry when full + Given an audit trail with max_entries 3 + When I add 4 step_allowed entries to the audit trail + Then the audit trail should have 3 entries + And the audit trail allowed_count should be 3 + + # ---- Edge cases ---- + + Scenario: Budget blocks when tool_budget is zero + Given guardrails with tool_budget 0.0 + When I check tool budget for cost 0.01 + Then the budget check should be denied + + Scenario: Budget allows zero cost when tool_budget is zero + Given guardrails with tool_budget 0.0 + When I check tool budget for cost 0 + Then the budget check should be allowed + + Scenario: Case-insensitive confirmation matching + Given guardrails with required confirmations "Deploy,DELETE" + When I check confirmation for operation "deploy" + Then confirmation should be required + + Scenario: Case-insensitive confirmation matching uppercase query + Given guardrails with required confirmations "deploy" + When I check confirmation for operation "DEPLOY" + Then confirmation should be required + + Scenario: Service rejects oversized metadata entries + Given an autonomy guardrail service + When I try to load metadata with oversized audit trail for plan "plan-big" + Then a guardrails validation error should be raised + + Scenario: Service rejects oversized confirmations in metadata + Given an autonomy guardrail service + When I try to load metadata with oversized confirmations for plan "plan-big2" + Then a guardrails validation error should be raised + + Scenario: Metadata round-trip preserves guardrail state + Given an autonomy guardrail service with guardrails for plan "plan-rt" + When I check step limit via service for plan "plan-rt" at step 3 + And I serialize and restore metadata for plan "plan-rt" + Then the restored guardrails for plan "plan-rt" should match the original + + # ---- start_time validation ---- + + Scenario: start_time accepts valid ISO-8601 timestamp + When I create guardrails with a valid start_time + Then the guardrails start_time should be set + + Scenario: start_time rejects malformed string + When I try to create guardrails with start_time "not-a-date" + Then a guardrails validation error should be raised + And the guardrails error should mention "start_time" + + Scenario: start_time accepts None + When I create autonomy guardrails with defaults + Then the guardrails start_time should be None + + Scenario: check_wall_clock handles malformed start_time gracefully + Given guardrails with a malformed start_time bypassing validation + When I check wall-clock time + Then the wall-clock check should be denied + + # ---- Retry-per-failure checks ---- + + Scenario: Retry limit allows within bounds + Given guardrails with retry limit 3 + When I check retries per failure with 2 current retries + Then the retry check should be allowed + + Scenario: Retry limit blocks at max + Given guardrails with retry limit 3 + When I check retries per failure with 3 current retries + Then the retry check should be denied + + Scenario: Retry limit blocks above max + Given guardrails with retry limit 2 + When I check retries per failure with 5 current retries + Then the retry check should be denied + + Scenario: Unlimited retries always allowed + Given guardrails with no retry limit + When I check retries per failure with 1000 current retries + Then the retry check should be allowed + + Scenario: Service retry check records audit trail + Given an autonomy guardrail service with retry-limit guardrails for plan "plan-rl" + When I check retries via service for plan "plan-rl" with 1 retries + Then the audit trail for plan "plan-rl" should have 1 entry + + Scenario: Service retry check blocks and records audit trail + Given an autonomy guardrail service with retry-limit guardrails for plan "plan-rl2" + When I check retries via service for plan "plan-rl2" with 5 retries + Then the retry service check should return False + And the audit trail for plan "plan-rl2" should have 1 entry + + Scenario: Successive retries stop after reaching max_retries_per_failure + Given a guardrail service with max_retries_per_failure 3 for plan "retry-stop" + When I check retries for plan "retry-stop" incrementing from 0 to 4 + Then retry checks 0, 1, 2 should be allowed + And retry checks 3, 4 should be denied + And the audit trail for plan "retry-stop" should have 5 entry + + # ---- PlanExecutor guardrail integration ---- + + Scenario: PlanExecutor enforces step limit during stub execution + Given a plan executor with guardrail service and step limit 2 + And a plan in execute phase with 3 decisions + When I run execute on the plan + Then execution should fail with a guardrail step limit error + + Scenario: PlanExecutor enforces wall-clock limit during execution + Given a plan executor with guardrail service and expired wall-clock + And a plan in execute phase with 1 decisions + When I run execute on the plan + Then execution should fail with a guardrail wall-clock error + + Scenario: PlanExecutor runs without guardrails when service is None + Given a plan executor without guardrail service + And a plan in execute phase with 2 decisions + When I run execute on the plan + Then execution should succeed + + Scenario: PlanExecutor marks start_time on first execution + Given a plan executor with guardrail service and no wall-clock limit + And a plan in execute phase with 1 decisions + When I run execute on the plan + Then the guardrails start_time should have been set + + # ---- Additional validation edge cases for coverage ---- + + Scenario: Guardrails reject invalid max_tool_calls_per_invocation + When I try to create actor limits with max_tool_calls_per_invocation 0 + Then a guardrails validation error should be raised + And the guardrails error should mention "max_tool_calls_per_invocation" + + Scenario: Guardrails reject negative max_retries_per_failure + When I try to create actor limits with max_retries_per_failure -1 + Then a guardrails validation error should be raised + And the guardrails error should mention "max_retries_per_failure" + + Scenario: Guardrails reject non-positive max_wall_clock_seconds + When I try to create guardrails with max_wall_clock_seconds 0 + Then a guardrails validation error should be raised + And the guardrails error should mention "max_wall_clock_seconds" + + Scenario: Audit trail evicts denied entry when full + Given an audit trail with max_entries 2 + When I add a budget_blocked entry to the audit trail + And I add a step_allowed entry to the audit trail + And I add a step_allowed entry to the audit trail + Then the audit trail should have 2 entries + And the audit trail denied_count should be 0 + And the audit trail allowed_count should be 2 diff --git a/features/steps/autonomy_guardrails_steps.py b/features/steps/autonomy_guardrails_steps.py new file mode 100644 index 000000000..e99b8b360 --- /dev/null +++ b/features/steps/autonomy_guardrails_steps.py @@ -0,0 +1,1050 @@ +"""Step definitions for autonomy guardrails and audit trail scenarios.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.autonomy_guardrail_service import ( + _MAX_CONFIRMATIONS, + _MAX_METADATA_ENTRIES, + AutonomyGuardrailService, +) +from cleveragents.application.services.plan_executor import ( + PlanExecutor, +) +from cleveragents.core.exceptions import PlanError +from cleveragents.domain.models.core.autonomy_guardrails import ( + ActorLimits, + AutonomyGuardrails, + GuardrailAuditEntry, + GuardrailAuditTrail, + GuardrailEventType, + GuardrailResult, +) +from cleveragents.domain.models.core.plan import ( + PlanPhase, + PlanTimestamps, + ProcessingState, +) + +# ---- Model creation steps ---- + + +@when( + "I create autonomy guardrails with max_steps {max_steps:d}" + " and tool_budget {budget:g}" +) +def step_create_guardrails( + context: Context, + max_steps: int, + budget: float, +) -> None: + context.guardrails = AutonomyGuardrails( + max_steps=max_steps, + tool_budget=budget, + ) + + +@when("I create autonomy guardrails with defaults") +def step_create_default_guardrails(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when("I try to create guardrails with max_steps {max_steps:d}") +def step_try_create_bad_max_steps( + context: Context, + max_steps: int, +) -> None: + try: + AutonomyGuardrails(max_steps=max_steps) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@when("I try to create guardrails with tool_budget {budget:g}") +def step_try_create_bad_budget( + context: Context, + budget: float, +) -> None: + try: + AutonomyGuardrails(tool_budget=budget) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@then("the guardrails should have max_steps {value}") +def step_check_max_steps(context: Context, value: str) -> None: + if value == "None": + assert context.guardrails.max_steps is None + else: + assert context.guardrails.max_steps == int(value) + + +@then("the guardrails should have tool_budget {value}") +def step_check_tool_budget(context: Context, value: str) -> None: + if value == "None": + assert context.guardrails.tool_budget is None + else: + assert context.guardrails.tool_budget == float(value) + + +@then("the guardrails step_count should be {count:d}") +def step_check_step_count(context: Context, count: int) -> None: + assert context.guardrails.step_count == count + + +@then("the guardrails budget_spent should be {amount:g}") +def step_check_budget_spent(context: Context, amount: float) -> None: + assert context.guardrails.budget_spent == amount + + +@then("the guardrails should have empty required_confirmations") +def step_check_empty_confirmations(context: Context) -> None: + assert context.guardrails.required_confirmations == [] + + +@then("a guardrails validation error should be raised") +def step_check_validation_error(context: Context) -> None: + assert context.guardrails_error is not None + + +@then('the guardrails error should mention "{text}"') +def step_check_error_message(context: Context, text: str) -> None: + assert text in str(context.guardrails_error) + + +# ---- Step limit checks ---- + + +@given("guardrails with max_steps {max_steps:d}") +def step_given_guardrails_max_steps( + context: Context, + max_steps: int, +) -> None: + context.guardrails = AutonomyGuardrails(max_steps=max_steps) + + +@given("guardrails with no step limit") +def step_given_no_step_limit(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when("I check step limit at step {step:d}") +def step_check_step_limit(context: Context, step: int) -> None: + context.guardrails.step_count = step + allowed, reason = context.guardrails.check_step_limit() + context.step_allowed = allowed + context.step_reason = reason + + +@then("the step check should be allowed") +def step_check_allowed(context: Context) -> None: + assert context.step_allowed is True + + +@then("the step check should be denied") +def step_check_denied(context: Context) -> None: + assert context.step_allowed is False + + +# ---- Budget checks ---- + + +@given("guardrails with tool_budget {budget:g}") +def step_given_guardrails_budget( + context: Context, + budget: float, +) -> None: + context.guardrails = AutonomyGuardrails(tool_budget=budget) + + +@given("guardrails with tool_budget {budget:g} and budget_spent {spent:g}") +def step_given_guardrails_budget_spent( + context: Context, + budget: float, + spent: float, +) -> None: + context.guardrails = AutonomyGuardrails( + tool_budget=budget, + budget_spent=spent, + ) + + +@given("guardrails with no budget limit") +def step_given_no_budget_limit(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when("I check tool budget for cost {cost:g}") +def step_check_budget(context: Context, cost: float) -> None: + allowed, reason = context.guardrails.check_tool_budget(cost) + context.budget_allowed = allowed + context.budget_reason = reason + + +@when("I try to check tool budget for cost {cost:g}") +def step_try_check_budget_negative( + context: Context, + cost: float, +) -> None: + try: + context.guardrails.check_tool_budget(cost) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@then("the budget check should be allowed") +def step_budget_allowed(context: Context) -> None: + assert context.budget_allowed is True + + +@then("the budget check should be denied") +def step_budget_denied(context: Context) -> None: + assert context.budget_allowed is False + + +# ---- Confirmation checks ---- + + +@given('guardrails with required confirmations "{confirmations}"') +def step_given_confirmations( + context: Context, + confirmations: str, +) -> None: + context.guardrails = AutonomyGuardrails( + required_confirmations=confirmations.split(","), + ) + + +@given("guardrails with no required confirmations") +def step_given_no_confirmations(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when('I check confirmation for operation "{operation}"') +def step_check_confirmation(context: Context, operation: str) -> None: + required, reason = context.guardrails.check_confirmation_required(operation) + context.confirmation_required = required + context.confirmation_reason = reason + + +@then("confirmation should be required") +def step_confirmation_required(context: Context) -> None: + assert context.confirmation_required is True + + +@then("confirmation should not be required") +def step_confirmation_not_required(context: Context) -> None: + assert context.confirmation_required is False + + +# ---- Step counter and budget tracking ---- + + +@when("I increment the step counter {count:d} times") +def step_increment_counter(context: Context, count: int) -> None: + for _ in range(count): + context.guardrails.increment_step() + + +@then("the step_count should be {count:d}") +def step_verify_step_count(context: Context, count: int) -> None: + assert context.guardrails.step_count == count + + +@when("I record a tool cost of {cost:g}") +def step_record_cost(context: Context, cost: float) -> None: + context.guardrails.record_cost(cost) + + +@then("the budget_spent should be {amount:g}") +def step_verify_budget_spent(context: Context, amount: float) -> None: + assert abs(context.guardrails.budget_spent - amount) < 0.001 + + +@when("I try to record a negative cost") +def step_try_record_negative_cost(context: Context) -> None: + try: + context.guardrails.record_cost(-5.0) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +# ---- Audit trail ---- + + +@given("an empty audit trail") +def step_given_empty_trail(context: Context) -> None: + context.audit_trail = GuardrailAuditTrail() + + +@when("I add a step_allowed entry to the audit trail") +def step_add_allowed_entry(context: Context) -> None: + entry = GuardrailAuditEntry( + event_type=GuardrailEventType.STEP_ALLOWED, + guard_name="step_limit", + result=GuardrailResult.ALLOWED, + ) + context.audit_trail.add_entry(entry) + + +@when("I add a budget_blocked entry to the audit trail") +def step_add_blocked_entry(context: Context) -> None: + entry = GuardrailAuditEntry( + event_type=GuardrailEventType.BUDGET_BLOCKED, + guard_name="tool_budget", + result=GuardrailResult.DENIED, + reason="Budget exceeded", + ) + context.audit_trail.add_entry(entry) + + +@then("the audit trail should have {count:d} entries") +def step_check_trail_count(context: Context, count: int) -> None: + trail = getattr(context, "audit_trail", None) + if trail is None: + trail = context.service_trail + assert len(trail.entries) == count + + +@then("the audit trail allowed_count should be {count:d}") +def step_check_allowed_count(context: Context, count: int) -> None: + assert context.audit_trail.allowed_count == count + + +@then("the audit trail denied_count should be {count:d}") +def step_check_denied_count(context: Context, count: int) -> None: + assert context.audit_trail.denied_count == count + + +@when( + "I create an audit entry with event_type {event_type} and guard_name {guard_name}" +) +def step_create_entry( + context: Context, + event_type: str, + guard_name: str, +) -> None: + context.audit_entry = GuardrailAuditEntry( + event_type=GuardrailEventType(event_type), + guard_name=guard_name, + result=GuardrailResult.DENIED, + reason="Test reason", + ) + + +@then('the audit entry event_type should be "{value}"') +def step_check_entry_event_type(context: Context, value: str) -> None: + assert context.audit_entry.event_type.value == value + + +@then('the audit entry guard_name should be "{value}"') +def step_check_entry_guard_name(context: Context, value: str) -> None: + assert context.audit_entry.guard_name == value + + +@then('the audit entry result should be "{value}"') +def step_check_entry_result(context: Context, value: str) -> None: + assert context.audit_entry.result.value == value + + +@then("the audit entry should have a timestamp") +def step_check_entry_timestamp(context: Context) -> None: + assert context.audit_entry.timestamp is not None + assert len(context.audit_entry.timestamp) > 0 + + +# ---- Service steps ---- + + +@given("an autonomy guardrail service") +def step_given_service(context: Context) -> None: + context.guardrail_service = AutonomyGuardrailService() + + +@when('I configure guardrails for plan "{plan_id}" with max_steps {max_steps:d}') +def step_configure_guardrails( + context: Context, + plan_id: str, + max_steps: int, +) -> None: + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails(max_steps=max_steps), + ) + + +@then('the service should return guardrails for plan "{plan_id}"') +def step_service_has_guardrails( + context: Context, + plan_id: str, +) -> None: + g = context.guardrail_service.get_guardrails(plan_id) + assert g is not None + context.service_guardrails = g + + +@then('the service should return None for plan "{plan_id}"') +def step_service_no_guardrails( + context: Context, + plan_id: str, +) -> None: + g = context.guardrail_service.get_guardrails(plan_id) + assert g is None + + +@then("the service guardrails should have max_steps {value:d}") +def step_service_max_steps(context: Context, value: int) -> None: + assert context.service_guardrails.max_steps == value + + +@given('an autonomy guardrail service with guardrails for plan "{plan_id}"') +def step_given_service_with_guardrails( + context: Context, + plan_id: str, +) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails(max_steps=10), + ) + + +@given('an autonomy guardrail service with budget guardrails for plan "{plan_id}"') +def step_given_service_with_budget( + context: Context, + plan_id: str, +) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails(tool_budget=100.0), + ) + + +@given( + 'an autonomy guardrail service with confirmation guardrails for plan "{plan_id}"' +) +def step_given_service_with_confirmations( + context: Context, + plan_id: str, +) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails(required_confirmations=["deploy", "delete"]), + ) + + +@when('I check step limit via service for plan "{plan_id}" at step {step:d}') +def step_service_check_step( + context: Context, + plan_id: str, + step: int, +) -> None: + context.step_result = context.guardrail_service.check_step_limit(plan_id, step) + + +@when('I check tool budget via service for plan "{plan_id}" with cost {cost:g}') +def step_service_check_budget( + context: Context, + plan_id: str, + cost: float, +) -> None: + context.budget_result = context.guardrail_service.check_tool_budget(plan_id, cost) + + +@when('I check confirmation via service for plan "{plan_id}" operation "{operation}"') +def step_service_check_confirmation( + context: Context, + plan_id: str, + operation: str, +) -> None: + context.confirmation_result = context.guardrail_service.check_confirmation_required( + plan_id, operation + ) + + +@then('the audit trail for plan "{plan_id}" should have {count:d} entry') +def step_service_trail_count( + context: Context, + plan_id: str, + count: int, +) -> None: + trail = context.guardrail_service.get_audit_trail(plan_id) + context.service_trail = trail + assert len(trail.entries) == count + + +@then("the service guardrails budget_spent should be {amount:g}") +def step_service_budget_spent(context: Context, amount: float) -> None: + # Use the plan_id stored by the service budget step (plan-003) + guardrails = context.guardrail_service.get_guardrails("plan-003") + assert guardrails is not None, "No budget guardrails found for plan-003" + assert abs(guardrails.budget_spent - amount) < 0.001 + + +@then('the metadata for plan "{plan_id}" should contain the audit trail') +def step_service_metadata(context: Context, plan_id: str) -> None: + metadata = context.guardrail_service.get_audit_trail_as_metadata(plan_id) + assert "guardrail_audit_trail" in metadata + assert "autonomy_guardrails" in metadata + + +@when('I load guardrails from metadata for plan "{plan_id}"') +def step_load_metadata(context: Context, plan_id: str) -> None: + metadata = { + "autonomy_guardrails": { + "max_steps": 20, + "tool_budget": 50.0, + "required_confirmations": [], + "step_count": 0, + "budget_spent": 0.0, + }, + "guardrail_audit_trail": {"entries": []}, + } + context.guardrail_service.load_from_metadata(plan_id, metadata) + + +@when('I remove plan "{plan_id}" from the service') +def step_remove_plan(context: Context, plan_id: str) -> None: + context.guardrail_service.remove_plan(plan_id) + + +@then("the step limit check should return True") +def step_check_returns_true(context: Context) -> None: + assert context.step_result is True + + +@then("the tool budget check should return True") +def step_budget_returns_true(context: Context) -> None: + assert context.budget_result is True + + +@then("the confirmation check should return False") +def step_confirmation_returns_false(context: Context) -> None: + assert context.confirmation_result is False + + +@when("I try to configure guardrails with empty plan_id") +def step_try_empty_plan_id(context: Context) -> None: + try: + context.guardrail_service.configure_guardrails( + "", + AutonomyGuardrails(), + ) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@then("GuardrailEventType should have at least {count:d} members") +def step_check_event_type_members( + context: Context, + count: int, +) -> None: + assert len(GuardrailEventType) >= count + + +@then("GuardrailResult should have exactly {count:d} members") +def step_check_result_members(context: Context, count: int) -> None: + assert len(GuardrailResult) == count + + +# ---- Wall-clock time steps ---- + + +@given("guardrails with no wall-clock limit") +def step_given_no_wall_clock(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@given("guardrails with wall-clock limit {seconds:g} seconds started recently") +def step_given_wall_clock_recent(context: Context, seconds: float) -> None: + context.guardrails = AutonomyGuardrails(max_wall_clock_seconds=seconds) + context.guardrails.mark_started() + + +@given("guardrails with wall-clock limit that has already expired") +def step_given_wall_clock_expired(context: Context) -> None: + context.guardrails = AutonomyGuardrails(max_wall_clock_seconds=0.001) + # Set start_time far in the past to guarantee expiry + context.guardrails.start_time = ( + datetime.now(tz=UTC) - timedelta(seconds=10) + ).isoformat() + + +@when("I check wall-clock time") +def step_check_wall_clock(context: Context) -> None: + allowed, reason = context.guardrails.check_wall_clock() + context.wall_clock_allowed = allowed + context.wall_clock_reason = reason + + +@then("the wall-clock check should be allowed") +def step_wall_clock_allowed(context: Context) -> None: + assert context.wall_clock_allowed is True + + +@then("the wall-clock check should be denied") +def step_wall_clock_denied(context: Context) -> None: + assert context.wall_clock_allowed is False + + +@given('an autonomy guardrail service with wall-clock guardrails for plan "{plan_id}"') +def step_given_service_wall_clock(context: Context, plan_id: str) -> None: + context.guardrail_service = AutonomyGuardrailService() + g = AutonomyGuardrails(max_wall_clock_seconds=600.0) + g.mark_started() + context.guardrail_service.configure_guardrails(plan_id, g) + + +@when('I check wall-clock via service for plan "{plan_id}"') +def step_service_check_wall_clock(context: Context, plan_id: str) -> None: + context.wall_clock_result = context.guardrail_service.check_wall_clock(plan_id) + + +# ---- Actor tool-call limit steps ---- + + +@given("guardrails with actor tool-call limit {limit:d}") +def step_given_actor_limit(context: Context, limit: int) -> None: + context.guardrails = AutonomyGuardrails( + actor_limits=ActorLimits(max_tool_calls_per_invocation=limit), + ) + + +@given("guardrails with no actor tool-call limit") +def step_given_no_actor_limit(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when("I check actor tool calls with {calls:d} current calls") +def step_check_actor_calls(context: Context, calls: int) -> None: + allowed, reason = context.guardrails.check_actor_tool_calls(calls) + context.actor_allowed = allowed + context.actor_reason = reason + + +@then("the actor tool-call check should be allowed") +def step_actor_allowed(context: Context) -> None: + assert context.actor_allowed is True + + +@then("the actor tool-call check should be denied") +def step_actor_denied(context: Context) -> None: + assert context.actor_allowed is False + + +@given('an autonomy guardrail service with actor-limit guardrails for plan "{plan_id}"') +def step_given_service_actor_limit(context: Context, plan_id: str) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails( + actor_limits=ActorLimits(max_tool_calls_per_invocation=10), + ), + ) + + +@when('I check actor tool calls via service for plan "{plan_id}" with {calls:d} calls') +def step_service_check_actor_calls( + context: Context, + plan_id: str, + calls: int, +) -> None: + context.actor_result = context.guardrail_service.check_actor_tool_calls( + plan_id, calls + ) + + +# ---- Bounded audit trail eviction steps ---- + + +@given("an audit trail with max_entries {max_entries:d}") +def step_given_bounded_trail(context: Context, max_entries: int) -> None: + context.audit_trail = GuardrailAuditTrail(max_entries=max_entries) + + +@when("I add {count:d} step_allowed entries to the audit trail") +def step_add_many_allowed_entries(context: Context, count: int) -> None: + for _ in range(count): + entry = GuardrailAuditEntry( + event_type=GuardrailEventType.STEP_ALLOWED, + guard_name="step_limit", + result=GuardrailResult.ALLOWED, + ) + context.audit_trail.add_entry(entry) + + +# ---- Edge-case steps ---- + + +@when('I try to load metadata with oversized audit trail for plan "{plan_id}"') +def step_try_load_oversized_trail(context: Context, plan_id: str) -> None: + metadata = { + "guardrail_audit_trail": { + "entries": [{}] * (_MAX_METADATA_ENTRIES + 1), + }, + } + try: + context.guardrail_service.load_from_metadata(plan_id, metadata) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@when('I try to load metadata with oversized confirmations for plan "{plan_id}"') +def step_try_load_oversized_confirmations( + context: Context, + plan_id: str, +) -> None: + metadata = { + "autonomy_guardrails": { + "max_steps": 10, + "tool_budget": 50.0, + "required_confirmations": ["op"] * (_MAX_CONFIRMATIONS + 1), + "step_count": 0, + "budget_spent": 0.0, + }, + } + try: + context.guardrail_service.load_from_metadata(plan_id, metadata) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@when('I serialize and restore metadata for plan "{plan_id}"') +def step_round_trip_metadata(context: Context, plan_id: str) -> None: + metadata = context.guardrail_service.get_audit_trail_as_metadata(plan_id) + context.round_trip_metadata = metadata + # Create a fresh service and restore + context.restored_service = AutonomyGuardrailService() + context.restored_service.load_from_metadata(plan_id, metadata) + + +@then('the restored guardrails for plan "{plan_id}" should match the original') +def step_verify_round_trip(context: Context, plan_id: str) -> None: + original = context.guardrail_service.get_guardrails(plan_id) + restored = context.restored_service.get_guardrails(plan_id) + assert original is not None + assert restored is not None + assert restored.max_steps == original.max_steps + assert restored.tool_budget == original.tool_budget + assert restored.required_confirmations == original.required_confirmations + assert restored.step_count == original.step_count + # Check audit trail was also restored + original_trail = context.guardrail_service.get_audit_trail(plan_id) + restored_trail = context.restored_service.get_audit_trail(plan_id) + assert len(restored_trail.entries) == len(original_trail.entries) + + +# ---- start_time validation steps ---- + + +@when("I create guardrails with a valid start_time") +def step_create_guardrails_valid_start_time(context: Context) -> None: + context.guardrails = AutonomyGuardrails( + start_time=datetime.now(tz=UTC).isoformat(), + ) + + +@when('I try to create guardrails with start_time "{value}"') +def step_try_create_bad_start_time(context: Context, value: str) -> None: + try: + AutonomyGuardrails(start_time=value) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@then("the guardrails start_time should be set") +def step_check_start_time_set(context: Context) -> None: + assert context.guardrails.start_time is not None + assert len(context.guardrails.start_time) > 0 + + +@then("the guardrails start_time should be None") +def step_check_start_time_none(context: Context) -> None: + assert context.guardrails.start_time is None + + +@given("guardrails with a malformed start_time bypassing validation") +def step_given_malformed_start_time(context: Context) -> None: + context.guardrails = AutonomyGuardrails(max_wall_clock_seconds=60.0) + # Bypass Pydantic validation by setting directly via object.__setattr__ + object.__setattr__(context.guardrails, "start_time", "not-a-date") + + +# ---- Retry-per-failure check steps ---- + + +@given("guardrails with retry limit {limit:d}") +def step_given_retry_limit(context: Context, limit: int) -> None: + context.guardrails = AutonomyGuardrails( + actor_limits=ActorLimits(max_retries_per_failure=limit), + ) + + +@given("guardrails with no retry limit") +def step_given_no_retry_limit(context: Context) -> None: + context.guardrails = AutonomyGuardrails() + + +@when("I check retries per failure with {retries:d} current retries") +def step_check_retries(context: Context, retries: int) -> None: + allowed, reason = context.guardrails.check_retries_per_failure(retries) + context.retry_allowed = allowed + context.retry_reason = reason + + +@then("the retry check should be allowed") +def step_retry_allowed(context: Context) -> None: + assert context.retry_allowed is True + + +@then("the retry check should be denied") +def step_retry_denied(context: Context) -> None: + assert context.retry_allowed is False + + +@given('an autonomy guardrail service with retry-limit guardrails for plan "{plan_id}"') +def step_given_service_retry_limit(context: Context, plan_id: str) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails( + actor_limits=ActorLimits(max_retries_per_failure=3), + ), + ) + + +@when('I check retries via service for plan "{plan_id}" with {retries:d} retries') +def step_service_check_retries( + context: Context, + plan_id: str, + retries: int, +) -> None: + context.retry_result = context.guardrail_service.check_retries_per_failure( + plan_id, retries + ) + + +@then("the retry service check should return False") +def step_retry_service_returns_false(context: Context) -> None: + assert context.retry_result is False + + +@given( + 'a guardrail service with max_retries_per_failure {limit:d} for plan "{plan_id}"' +) +def step_given_service_custom_retry_limit( + context: Context, limit: int, plan_id: str +) -> None: + context.guardrail_service = AutonomyGuardrailService() + context.guardrail_service.configure_guardrails( + plan_id, + AutonomyGuardrails( + actor_limits=ActorLimits(max_retries_per_failure=limit), + ), + ) + + +@when('I check retries for plan "{plan_id}" incrementing from {start:d} to {end:d}') +def step_check_retries_incrementing( + context: Context, plan_id: str, start: int, end: int +) -> None: + context.retry_results = [] + for i in range(start, end + 1): + allowed = context.guardrail_service.check_retries_per_failure(plan_id, i) + context.retry_results.append((i, allowed)) + + +@then("retry checks {allowed_csv} should be allowed") +def step_retry_checks_allowed(context: Context, allowed_csv: str) -> None: + expected = [int(x.strip()) for x in allowed_csv.split(",")] + results_map = dict(context.retry_results) + for idx in expected: + assert results_map[idx] is True, ( + f"Expected retry {idx} to be allowed, but it was denied" + ) + + +@then("retry checks {denied_csv} should be denied") +def step_retry_checks_denied(context: Context, denied_csv: str) -> None: + expected = [int(x.strip()) for x in denied_csv.split(",")] + results_map = dict(context.retry_results) + for idx in expected: + assert results_map[idx] is False, ( + f"Expected retry {idx} to be denied, but it was allowed" + ) + + +# ---- PlanExecutor guardrail integration helpers ---- + + +def _make_guardrail_mock_plan( + phase: PlanPhase = PlanPhase.EXECUTE, + state: ProcessingState = ProcessingState.QUEUED, + definition_of_done: str = "Step A\nStep B", + decision_root_id: str = "root-001", +) -> MagicMock: + """Create a mock plan suitable for PlanExecutor tests.""" + plan = MagicMock() + plan.phase = phase + plan.state = state + plan.definition_of_done = definition_of_done + plan.decision_root_id = decision_root_id + plan.timestamps = PlanTimestamps() + plan.error_details = {} + plan.changeset_id = None + plan.sandbox_refs = [] + plan.read_only = False + plan.invariants = [] + return plan + + +def _make_guardrail_mock_lifecycle(plan: MagicMock) -> MagicMock: + """Create a mock lifecycle service for PlanExecutor tests.""" + lcs = MagicMock() + lcs.get_plan.return_value = plan + lcs.start_strategize = MagicMock() + lcs.complete_strategize = MagicMock() + lcs.fail_strategize = MagicMock() + lcs.start_execute = MagicMock() + lcs.complete_execute = MagicMock() + lcs.fail_execute = MagicMock() + lcs._commit_plan = MagicMock() + return lcs + + +# ---- PlanExecutor guardrail integration steps ---- + + +@given("a plan executor with guardrail service and step limit {limit:d}") +def step_given_executor_with_step_limit(context: Context, limit: int) -> None: + service = AutonomyGuardrailService() + service.configure_guardrails( + "grd-plan-001", + AutonomyGuardrails(max_steps=limit), + ) + context.guardrail_plan_id = "grd-plan-001" + context.guardrail_exec_service = service + + +@given("a plan executor with guardrail service and expired wall-clock") +def step_given_executor_with_expired_wall_clock(context: Context) -> None: + service = AutonomyGuardrailService() + g = AutonomyGuardrails(max_wall_clock_seconds=0.001) + g.start_time = (datetime.now(tz=UTC) - timedelta(seconds=10)).isoformat() + service.configure_guardrails("grd-plan-002", g) + context.guardrail_plan_id = "grd-plan-002" + context.guardrail_exec_service = service + + +@given("a plan executor without guardrail service") +def step_given_executor_no_guardrails(context: Context) -> None: + context.guardrail_plan_id = "grd-plan-003" + context.guardrail_exec_service = None + + +@given("a plan executor with guardrail service and no wall-clock limit") +def step_given_executor_no_wall_clock(context: Context) -> None: + service = AutonomyGuardrailService() + service.configure_guardrails( + "grd-plan-004", + AutonomyGuardrails(max_steps=100), + ) + context.guardrail_plan_id = "grd-plan-004" + context.guardrail_exec_service = service + + +@given("a plan in execute phase with {count:d} decisions") +def step_given_plan_in_execute_phase(context: Context, count: int) -> None: + steps = "\n".join(f"Step {chr(65 + i)}" for i in range(count)) + plan = _make_guardrail_mock_plan(definition_of_done=steps) + lifecycle = _make_guardrail_mock_lifecycle(plan) + context.guardrail_executor = PlanExecutor( + lifecycle_service=lifecycle, + guardrail_service=context.guardrail_exec_service, + ) + + +@when("I run execute on the plan") +def step_run_execute(context: Context) -> None: + try: + context.execute_result = context.guardrail_executor.run_execute( + context.guardrail_plan_id, + ) + context.execute_error = None + except Exception as exc: + context.execute_error = exc + context.execute_result = None + + +@then("execution should fail with a guardrail step limit error") +def step_check_step_limit_error(context: Context) -> None: + assert context.execute_error is not None + assert isinstance(context.execute_error, PlanError) + assert "step limit" in str(context.execute_error).lower() + + +@then("execution should fail with a guardrail wall-clock error") +def step_check_wall_clock_error(context: Context) -> None: + assert context.execute_error is not None + assert isinstance(context.execute_error, PlanError) + assert "wall-clock" in str(context.execute_error).lower() + + +@then("execution should succeed") +def step_check_execution_success(context: Context) -> None: + assert context.execute_error is None + assert context.execute_result is not None + + +@then("the guardrails start_time should have been set") +def step_check_start_time_was_set(context: Context) -> None: + service = context.guardrail_exec_service + assert service is not None + guardrails = service.get_guardrails(context.guardrail_plan_id) + assert guardrails is not None + assert guardrails.start_time is not None + + +# ---- Additional validation edge-case steps ---- + + +@when("I try to create actor limits with max_tool_calls_per_invocation {value:d}") +def step_try_bad_max_tool_calls(context: Context, value: int) -> None: + try: + ActorLimits(max_tool_calls_per_invocation=value) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@when("I try to create actor limits with max_retries_per_failure {value:d}") +def step_try_bad_max_retries(context: Context, value: int) -> None: + try: + ActorLimits(max_retries_per_failure=value) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc + + +@when("I try to create guardrails with max_wall_clock_seconds {value:d}") +def step_try_bad_wall_clock(context: Context, value: int) -> None: + try: + AutonomyGuardrails(max_wall_clock_seconds=float(value)) + context.guardrails_error = None + except Exception as exc: + context.guardrails_error = exc diff --git a/robot/autonomy_guardrails.robot b/robot/autonomy_guardrails.robot new file mode 100644 index 000000000..9aa86dde7 --- /dev/null +++ b/robot/autonomy_guardrails.robot @@ -0,0 +1,87 @@ +*** Settings *** +Documentation Smoke tests for autonomy guardrails and audit trail +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_autonomy_guardrails.py + +*** Test Cases *** +Guardrail Step Limit Allows Within Bounds + [Documentation] Step limit allows when under limit + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} step-allow cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} step-allow-ok + +Guardrail Step Limit Blocks At Max + [Documentation] Step limit blocks when at max steps + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} step-block cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} step-block-ok + +Guardrail Budget Allows Within Limit + [Documentation] Budget allows when within limit + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} budget-allow cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} budget-allow-ok + +Guardrail Budget Blocks Exceeded + [Documentation] Budget blocks when exceeded + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} budget-block cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} budget-block-ok + +Guardrail Confirmation Required + [Documentation] Confirmation required for listed operation + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} confirm-required cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} confirm-required-ok + +Guardrail Audit Trail Records Events + [Documentation] Audit trail records enforcement events + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} audit-trail cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} audit-trail-ok + +Guardrail Service Metadata Persistence + [Documentation] Service can serialize and restore from metadata + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} metadata cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} metadata-ok + +Guardrail Wall Clock Allows Within Limit + [Documentation] Wall-clock allows when within time limit + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} wall-clock-allow cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wall-clock-allow-ok + +Guardrail Wall Clock Blocks When Expired + [Documentation] Wall-clock blocks when time limit exceeded + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} wall-clock-block cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wall-clock-block-ok + +Guardrail Actor Tool Call Limit + [Documentation] Actor tool-call limit enforcement + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} actor-limit cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} actor-limit-ok + +Guardrail Audit Trail Eviction + [Documentation] Audit trail evicts oldest entries when full + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} eviction cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} eviction-ok + +Guardrail Zero Budget Edge Case + [Documentation] Zero budget blocks any positive cost + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} zero-budget cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} zero-budget-ok + +Guardrail CLI Flags Summary + [Documentation] Summary of autonomy guardrail features + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} summary cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} summary-ok diff --git a/robot/helper_autonomy_guardrails.py b/robot/helper_autonomy_guardrails.py new file mode 100644 index 000000000..b1f4783f7 --- /dev/null +++ b/robot/helper_autonomy_guardrails.py @@ -0,0 +1,220 @@ +"""Helper script for Robot Framework autonomy guardrails tests.""" + +from __future__ import annotations + +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +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, +) + + +def _test_step_allow() -> None: + """Step limit allows within bounds.""" + g = AutonomyGuardrails(max_steps=10) + g.step_count = 5 + allowed, _ = g.check_step_limit() + assert allowed is True + print("step-allow-ok") + + +def _test_step_block() -> None: + """Step limit blocks at max.""" + g = AutonomyGuardrails(max_steps=5) + g.step_count = 5 + allowed, reason = g.check_step_limit() + assert allowed is False + assert reason is not None + assert "Step limit" in reason + print("step-block-ok") + + +def _test_budget_allow() -> None: + """Budget allows within limit.""" + g = AutonomyGuardrails(tool_budget=100.0) + allowed, _ = g.check_tool_budget(50.0) + assert allowed is True + print("budget-allow-ok") + + +def _test_budget_block() -> None: + """Budget blocks when exceeded.""" + g = AutonomyGuardrails(tool_budget=10.0, budget_spent=8.0) + allowed, reason = g.check_tool_budget(5.0) + assert allowed is False + assert reason is not None + assert "Budget exceeded" in reason + print("budget-block-ok") + + +def _test_confirm_required() -> None: + """Confirmation required for listed operation.""" + g = AutonomyGuardrails(required_confirmations=["deploy", "delete"]) + required, reason = g.check_confirmation_required("deploy") + assert required is True + assert reason is not None + not_required, _ = g.check_confirmation_required("read") + assert not_required is False + print("confirm-required-ok") + + +def _test_audit_trail() -> None: + """Audit trail records enforcement events.""" + svc = AutonomyGuardrailService() + svc.configure_guardrails( + "test-plan", + AutonomyGuardrails(max_steps=5, tool_budget=50.0), + ) + svc.check_step_limit("test-plan", 3) + svc.check_tool_budget("test-plan", 10.0) + trail = svc.get_audit_trail("test-plan") + assert len(trail.entries) == 2 + assert trail.entries[0].guard_name == "step_limit" + assert trail.entries[1].guard_name == "tool_budget" + print("audit-trail-ok") + + +def _test_metadata() -> None: + """Service can serialize and restore from metadata.""" + svc = AutonomyGuardrailService() + svc.configure_guardrails( + "meta-plan", + AutonomyGuardrails(max_steps=10, tool_budget=25.0), + ) + svc.check_step_limit("meta-plan", 2) + + metadata = svc.get_audit_trail_as_metadata("meta-plan") + assert "guardrail_audit_trail" in metadata + assert "autonomy_guardrails" in metadata + + svc2 = AutonomyGuardrailService() + svc2.load_from_metadata("meta-plan", metadata) + g = svc2.get_guardrails("meta-plan") + assert g is not None + assert g.max_steps == 10 + print("metadata-ok") + + +def _test_summary() -> None: + """Summary of autonomy guardrail features.""" + g = AutonomyGuardrails( + max_steps=100, + tool_budget=500.0, + required_confirmations=["deploy", "delete"], + ) + assert g.max_steps == 100 + assert g.tool_budget == 500.0 + assert len(g.required_confirmations) == 2 + g.increment_step() + assert g.step_count == 1 + g.record_cost(10.0) + assert g.budget_spent == 10.0 + trail = GuardrailAuditTrail() + entry = GuardrailAuditEntry( + event_type=GuardrailEventType.STEP_ALLOWED, + guard_name="step_limit", + result=GuardrailResult.ALLOWED, + ) + trail.add_entry(entry) + assert len(trail.entries) == 1 + print("summary-ok") + + +def _test_wall_clock_allow() -> None: + """Wall-clock allows when within limit.""" + g = AutonomyGuardrails(max_wall_clock_seconds=600.0) + g.mark_started() + allowed, _ = g.check_wall_clock() + assert allowed is True + print("wall-clock-allow-ok") + + +def _test_wall_clock_block() -> None: + """Wall-clock blocks when expired.""" + g = AutonomyGuardrails(max_wall_clock_seconds=0.001) + g.start_time = (datetime.now(tz=UTC) - timedelta(seconds=10)).isoformat() + allowed, reason = g.check_wall_clock() + assert allowed is False + assert reason is not None + assert "Wall-clock" in reason + print("wall-clock-block-ok") + + +def _test_actor_limit() -> None: + """Actor tool-call limit enforcement.""" + g = AutonomyGuardrails( + actor_limits=ActorLimits(max_tool_calls_per_invocation=5), + ) + allowed, _ = g.check_actor_tool_calls(3) + assert allowed is True + blocked, reason = g.check_actor_tool_calls(5) + assert blocked is False + assert reason is not None + assert "Actor" in reason + print("actor-limit-ok") + + +def _test_eviction() -> None: + """Audit trail evicts oldest entries when full.""" + trail = GuardrailAuditTrail(max_entries=3) + for i in range(5): + entry = GuardrailAuditEntry( + event_type=GuardrailEventType.STEP_ALLOWED, + guard_name=f"step_{i}", + result=GuardrailResult.ALLOWED, + ) + trail.add_entry(entry) + assert len(trail.entries) == 3 + # Oldest two (step_0, step_1) should be evicted + assert trail.entries[0].guard_name == "step_2" + assert trail.allowed_count == 3 + print("eviction-ok") + + +def _test_zero_budget() -> None: + """Zero budget blocks any positive cost.""" + g = AutonomyGuardrails(tool_budget=0.0) + allowed, _ = g.check_tool_budget(0.0) + assert allowed is True + blocked, reason = g.check_tool_budget(0.01) + assert blocked is False + assert reason is not None + print("zero-budget-ok") + + +if __name__ == "__main__": + cmd = sys.argv[1] if len(sys.argv) > 1 else "summary" + dispatch: dict[str, Any] = { + "step-allow": _test_step_allow, + "step-block": _test_step_block, + "budget-allow": _test_budget_allow, + "budget-block": _test_budget_block, + "confirm-required": _test_confirm_required, + "audit-trail": _test_audit_trail, + "metadata": _test_metadata, + "wall-clock-allow": _test_wall_clock_allow, + "wall-clock-block": _test_wall_clock_block, + "actor-limit": _test_actor_limit, + "eviction": _test_eviction, + "zero-budget": _test_zero_budget, + "summary": _test_summary, + } + fn = dispatch.get(cmd) + if fn: + fn() + else: + print(f"Unknown command: {cmd}", file=sys.stderr) + sys.exit(1) diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 49a7422fc..08df9ad4c 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -11,6 +11,9 @@ from dependency_injector import containers, providers from cleveragents.actor.registry import ActorRegistry from cleveragents.application.reactive_registry_adapter import register_registry_agents from cleveragents.application.services.actor_service import ActorService +from cleveragents.application.services.autonomy_guardrail_service import ( + AutonomyGuardrailService, +) from cleveragents.application.services.context_service import ContextService from cleveragents.application.services.decision_service import DecisionService from cleveragents.application.services.plan_lifecycle_service import ( @@ -250,6 +253,11 @@ class Container(containers.DeclarativeContainer): database_url=database_url, ) + # Autonomy Guardrail Service - Singleton so all callers share state + autonomy_guardrail_service = providers.Singleton( + AutonomyGuardrailService, + ) + # Reactive routing stream_router = providers.Singleton(ReactiveStreamRouter) langgraph_bridge = providers.Singleton( diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 331cb96b8..8ed7916f4 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -3,6 +3,9 @@ Contains service classes that orchestrate business operations. """ +from cleveragents.application.services.autonomy_guardrail_service import ( + AutonomyGuardrailService, +) from cleveragents.application.services.config_service import ( ConfigEntry, ConfigLevel, @@ -82,6 +85,7 @@ __all__ = [ "ApplyValidationResult", "ApplyValidationSummary", "AttachmentScope", + "AutonomyGuardrailService", "BrokenReferenceRule", "ConfigEntry", "ConfigLevel", diff --git a/src/cleveragents/application/services/autonomy_guardrail_service.py b/src/cleveragents/application/services/autonomy_guardrail_service.py new file mode 100644 index 000000000..c1ff23358 --- /dev/null +++ b/src/cleveragents/application/services/autonomy_guardrail_service.py @@ -0,0 +1,524 @@ +"""Autonomy guardrail enforcement service. + +Provides a high-level service interface for checking autonomy +guardrails during plan execution and recording enforcement +events to the plan's audit trail. + +The service integrates with the plan metadata to persist audit +trail entries, providing a complete record of guardrail decisions. + +Thread-safe: all state mutations are protected by a reentrant lock +so the service can be shared across threads (e.g. concurrent plan +step evaluations). +""" + +from __future__ import annotations + +import logging +import threading +from typing import Any + +from cleveragents.domain.models.core.autonomy_guardrails import ( + AutonomyGuardrails, + GuardrailAuditEntry, + GuardrailAuditTrail, + GuardrailEventType, + GuardrailResult, +) + +logger = logging.getLogger(__name__) + +# Key used in plan metadata to store the audit trail +_AUDIT_TRAIL_KEY = "guardrail_audit_trail" + +# Key used in plan metadata to store guardrail state +_GUARDRAILS_KEY = "autonomy_guardrails" + +# Hard ceiling for entries accepted through load_from_metadata to +# prevent memory exhaustion from untrusted or corrupted payloads. +_MAX_METADATA_ENTRIES = 50_000 + +# Hard ceiling for required_confirmations list length. +_MAX_CONFIRMATIONS = 500 + + +class AutonomyGuardrailService: + """Service for enforcing autonomy guardrails during plan execution. + + Manages guardrail state and audit trail persistence via plan + metadata dictionaries. Each plan can have its own guardrail + configuration and audit trail stored in its metadata. + + All public methods are thread-safe. + """ + + def __init__(self) -> None: + """Initialize the autonomy guardrail service.""" + self._guardrails: dict[str, AutonomyGuardrails] = {} + self._audit_trails: dict[str, GuardrailAuditTrail] = {} + self._lock = threading.RLock() + + def configure_guardrails( + self, + plan_id: str, + guardrails: AutonomyGuardrails, + ) -> None: + """Configure guardrails for a plan. + + Args: + plan_id: The plan identifier. + guardrails: The guardrail configuration to apply. + """ + if not plan_id: + raise ValueError("plan_id must not be empty") + with self._lock: + self._guardrails[plan_id] = guardrails + if plan_id not in self._audit_trails: + self._audit_trails[plan_id] = GuardrailAuditTrail() + logger.debug("Configured guardrails for plan %s", plan_id) + + def get_guardrails(self, plan_id: str) -> AutonomyGuardrails | None: + """Retrieve the guardrail configuration for a plan. + + Args: + plan_id: The plan identifier. + + Returns: + The guardrail configuration, or None if not configured. + """ + with self._lock: + return self._guardrails.get(plan_id) + + def check_step_limit( + self, + plan_id: str, + current_step: int, + ) -> bool: + """Check if the plan has exceeded its maximum step limit. + + Args: + plan_id: The plan identifier. + current_step: The current step number. + + Returns: + True if the step is allowed, False if blocked. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return True + + guardrails.step_count = current_step + allowed, reason = guardrails.check_step_limit() + + event_type = ( + GuardrailEventType.STEP_ALLOWED + if allowed + else GuardrailEventType.STEP_BLOCKED + ) + result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="step_limit", + result=result, + reason=reason, + context={ + "current_step": current_step, + "max_steps": guardrails.max_steps, + }, + ) + self._record_enforcement_locked(plan_id, entry) + + if not allowed: + logger.warning( + "Step limit exceeded for plan %s: %s", + plan_id, + reason, + ) + + return allowed + + def check_tool_budget( + self, + plan_id: str, + tool_cost: float, + ) -> bool: + """Check if the plan's budget allows a tool invocation. + + When allowed, the cost is recorded **before** the audit entry + so that the trail reflects the post-deduction budget state. + + Args: + plan_id: The plan identifier. + tool_cost: The cost of the proposed tool invocation. + + Returns: + True if the invocation is within budget, False if blocked. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return True + + allowed, reason = guardrails.check_tool_budget(tool_cost) + + # Record cost *before* audit entry so context is accurate + if allowed: + guardrails.record_cost(tool_cost) + + event_type = ( + GuardrailEventType.BUDGET_ALLOWED + if allowed + else GuardrailEventType.BUDGET_BLOCKED + ) + result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="tool_budget", + result=result, + reason=reason, + context={ + "tool_cost": tool_cost, + "budget_spent": guardrails.budget_spent, + "tool_budget": guardrails.tool_budget, + }, + ) + self._record_enforcement_locked(plan_id, entry) + + if not allowed: + logger.warning( + "Tool budget exceeded for plan %s: %s", + plan_id, + reason, + ) + + return allowed + + def check_wall_clock(self, plan_id: str) -> bool: + """Check if the plan has exceeded its wall-clock time limit. + + Args: + plan_id: The plan identifier. + + Returns: + True if the plan is within the time limit, False if blocked. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return True + + allowed, reason = guardrails.check_wall_clock() + + event_type = ( + GuardrailEventType.TIME_ALLOWED + if allowed + else GuardrailEventType.TIME_BLOCKED + ) + result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="wall_clock", + result=result, + reason=reason, + context={ + "max_wall_clock_seconds": guardrails.max_wall_clock_seconds, + }, + ) + self._record_enforcement_locked(plan_id, entry) + + if not allowed: + logger.warning( + "Wall-clock limit exceeded for plan %s: %s", + plan_id, + reason, + ) + + return allowed + + def check_actor_tool_calls( + self, + plan_id: str, + current_calls: int, + ) -> bool: + """Check if the actor has exceeded its per-invocation tool call limit. + + Args: + plan_id: The plan identifier. + current_calls: Number of tool calls made so far. + + Returns: + True if allowed, False if blocked. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return True + + allowed, reason = guardrails.check_actor_tool_calls(current_calls) + + event_type = ( + GuardrailEventType.ACTOR_LIMIT_ALLOWED + if allowed + else GuardrailEventType.ACTOR_LIMIT_BLOCKED + ) + result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="actor_tool_calls", + result=result, + reason=reason, + context={ + "current_calls": current_calls, + "max_tool_calls_per_invocation": ( + guardrails.actor_limits.max_tool_calls_per_invocation + ), + }, + ) + self._record_enforcement_locked(plan_id, entry) + + if not allowed: + logger.warning( + "Actor tool-call limit exceeded for plan %s: %s", + plan_id, + reason, + ) + + return allowed + + def check_retries_per_failure( + self, + plan_id: str, + current_retries: int, + ) -> bool: + """Check if the retry count exceeds the per-failure retry limit. + + Args: + plan_id: The plan identifier. + current_retries: Number of retries attempted so far. + + Returns: + True if the retry is allowed, False if blocked. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return True + + allowed, reason = guardrails.check_retries_per_failure(current_retries) + + event_type = ( + GuardrailEventType.RETRY_ALLOWED + if allowed + else GuardrailEventType.RETRY_BLOCKED + ) + result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="retries_per_failure", + result=result, + reason=reason, + context={ + "current_retries": current_retries, + "max_retries_per_failure": ( + guardrails.actor_limits.max_retries_per_failure + ), + }, + ) + self._record_enforcement_locked(plan_id, entry) + + if not allowed: + logger.warning( + "Retry limit exceeded for plan %s: %s", + plan_id, + reason, + ) + + return allowed + + def check_confirmation_required( + self, + plan_id: str, + operation: str, + ) -> bool: + """Check if an operation requires human confirmation. + + Args: + plan_id: The plan identifier. + operation: The operation name to check. + + Returns: + True if confirmation is required, False otherwise. + """ + with self._lock: + guardrails = self._guardrails.get(plan_id) + if guardrails is None: + return False + + needs_confirmation, reason = guardrails.check_confirmation_required( + operation, + ) + + if needs_confirmation: + event_type = GuardrailEventType.CONFIRMATION_REQUIRED + result = GuardrailResult.DENIED + else: + event_type = GuardrailEventType.CONFIRMATION_GRANTED + result = GuardrailResult.ALLOWED + + entry = GuardrailAuditEntry( + event_type=event_type, + guard_name="confirmation", + result=result, + reason=reason, + context={"operation": operation}, + ) + self._record_enforcement_locked(plan_id, entry) + + if needs_confirmation: + logger.info( + "Confirmation required for plan %s operation '%s': %s", + plan_id, + operation, + reason, + ) + + return needs_confirmation + + def record_enforcement( + self, + plan_id: str, + entry: GuardrailAuditEntry, + ) -> None: + """Persist an audit entry to the plan's audit trail. + + Args: + plan_id: The plan identifier. + entry: The audit entry to record. + """ + with self._lock: + self._record_enforcement_locked(plan_id, entry) + + def get_audit_trail(self, plan_id: str) -> GuardrailAuditTrail: + """Retrieve the audit trail for a plan. + + Args: + plan_id: The plan identifier. + + Returns: + The audit trail for the plan. Returns an empty trail + if no events have been recorded. + """ + with self._lock: + return self._audit_trails.get(plan_id, GuardrailAuditTrail()) + + def get_audit_trail_as_metadata( + self, + plan_id: str, + ) -> dict[str, Any]: + """Serialize the audit trail for persistence in plan metadata. + + Args: + plan_id: The plan identifier. + + Returns: + A dictionary suitable for storing in plan metadata. + """ + with self._lock: + trail = self._audit_trails.get(plan_id, GuardrailAuditTrail()) + guardrails = self._guardrails.get(plan_id) + + metadata: dict[str, Any] = { + _AUDIT_TRAIL_KEY: trail.model_dump(), + } + if guardrails is not None: + metadata[_GUARDRAILS_KEY] = guardrails.model_dump() + return metadata + + def load_from_metadata( + self, + plan_id: str, + metadata: dict[str, Any], + ) -> None: + """Restore guardrail state from plan metadata. + + Applies size guards to prevent memory exhaustion from + oversized or malicious payloads. + + Args: + plan_id: The plan identifier. + metadata: The plan metadata dictionary. + + Raises: + ValueError: If the metadata contains oversized payloads. + """ + with self._lock: + if _GUARDRAILS_KEY in metadata: + raw_guardrails = metadata[_GUARDRAILS_KEY] + if isinstance(raw_guardrails, dict): + confirmations = raw_guardrails.get( + "required_confirmations", + [], + ) + if ( + isinstance(confirmations, list) + and len(confirmations) > _MAX_CONFIRMATIONS + ): + raise ValueError( + f"required_confirmations exceeds maximum of " + f"{_MAX_CONFIRMATIONS} entries" + ) + self._guardrails[plan_id] = AutonomyGuardrails.model_validate( + raw_guardrails, + ) + + if _AUDIT_TRAIL_KEY in metadata: + raw_trail = metadata[_AUDIT_TRAIL_KEY] + if isinstance(raw_trail, dict): + entries = raw_trail.get("entries", []) + if ( + isinstance(entries, list) + and len(entries) > _MAX_METADATA_ENTRIES + ): + raise ValueError( + f"Audit trail exceeds maximum of " + f"{_MAX_METADATA_ENTRIES} entries" + ) + self._audit_trails[plan_id] = GuardrailAuditTrail.model_validate( + raw_trail + ) + + def remove_plan(self, plan_id: str) -> None: + """Remove all guardrail state for a plan. + + Args: + plan_id: The plan identifier. + """ + with self._lock: + self._guardrails.pop(plan_id, None) + self._audit_trails.pop(plan_id, None) + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _record_enforcement_locked( + self, + plan_id: str, + entry: GuardrailAuditEntry, + ) -> None: + """Record an audit entry (caller must hold ``_lock``).""" + if plan_id not in self._audit_trails: + self._audit_trails[plan_id] = GuardrailAuditTrail() + self._audit_trails[plan_id].add_entry(entry) + logger.debug( + "Recorded guardrail event for plan %s: %s (%s)", + plan_id, + entry.event_type.value, + entry.result.value, + ) diff --git a/src/cleveragents/application/services/plan_executor.py b/src/cleveragents/application/services/plan_executor.py index 6da03a3e0..69c6f8147 100644 --- a/src/cleveragents/application/services/plan_executor.py +++ b/src/cleveragents/application/services/plan_executor.py @@ -19,6 +19,9 @@ import structlog from pydantic import BaseModel, ConfigDict, Field from ulid import ULID +from cleveragents.application.services.autonomy_guardrail_service import ( + AutonomyGuardrailService, +) from cleveragents.application.services.plan_execution_context import ( PlanExecutionContext, RuntimeExecuteActor, @@ -259,6 +262,7 @@ class PlanExecutor: execution_context: PlanExecutionContext | None = None, error_recovery_service: ErrorRecoveryService | None = None, checkpoint_manager: CheckpointManager | None = None, + guardrail_service: AutonomyGuardrailService | None = None, ) -> None: """Initialize the plan executor. @@ -272,6 +276,8 @@ class PlanExecutor: checkpoint_manager: Optional checkpoint manager for sandbox state snapshots during execute. When ``None``, checkpoint hooks are silently skipped. + guardrail_service: Optional autonomy guardrail service for + enforcing step limits, budgets, and wall-clock time. """ if lifecycle_service is None: raise ValidationError("lifecycle_service must not be None") @@ -281,6 +287,7 @@ class PlanExecutor: self._execution_context = execution_context self._error_recovery = error_recovery_service self._checkpoint_manager = checkpoint_manager + self._guardrail_service = guardrail_service self._strategize_actor = StrategizeStubActor() self._execute_actor = ExecuteStubActor() self._logger = logger.bind(service="plan_executor") @@ -520,8 +527,49 @@ class PlanExecutor: f"Plan {plan_id} has no decision tree. " "Strategize must complete before Execute." ) + self._enforce_guardrails(plan_id) return plan + def _enforce_guardrails(self, plan_id: str) -> None: + """Pre-flight guardrail checks before execution begins. + + Marks the execution start time (for wall-clock tracking) if + not already set, then checks wall-clock time and current step + limit. Raises ``PlanError`` if any guardrail blocks execution. + """ + if self._guardrail_service is None: + return + guardrails = self._guardrail_service.get_guardrails(plan_id) + if guardrails is not None and guardrails.start_time is None: + guardrails.mark_started() + if not self._guardrail_service.check_wall_clock(plan_id): + raise PlanError(f"Guardrail wall-clock limit exceeded for plan {plan_id}") + if guardrails is not None: + current_step = guardrails.step_count + if not self._guardrail_service.check_step_limit(plan_id, current_step): + raise PlanError( + f"Guardrail step limit already reached for plan {plan_id}" + ) + + def _enforce_guardrails_per_step(self, plan_id: str) -> None: + """Per-decision guardrail enforcement. + + Increments the step counter and checks step limit and + wall-clock time. Raises ``PlanError`` if blocked. + """ + if self._guardrail_service is None: + return + guardrails = self._guardrail_service.get_guardrails(plan_id) + if guardrails is None: + return # pragma: no cover - defensive + next_step = guardrails.step_count + 1 + if not self._guardrail_service.check_step_limit(plan_id, next_step): + raise PlanError( + f"Guardrail step limit reached for plan {plan_id} at step {next_step}" + ) + if not self._guardrail_service.check_wall_clock(plan_id): + raise PlanError(f"Guardrail wall-clock limit exceeded for plan {plan_id}") + def _run_execute_with_runtime( self, plan_id: str, @@ -542,6 +590,10 @@ class PlanExecutor: self._lifecycle.start_execute(plan_id) self._try_create_checkpoint(plan_id, "pre_execute") try: + # Enforce per-step guardrails for each decision + for _decision in decisions: + self._enforce_guardrails_per_step(plan_id) + result = runtime_actor.execute( decisions=decisions, stream_callback=stream_callback ) @@ -600,6 +652,10 @@ class PlanExecutor: for attempt in range(max_attempts): try: + # Enforce per-step guardrails for each decision + for _decision in decisions: + self._enforce_guardrails_per_step(plan_id) + result = self._execute_actor.execute( plan_id=plan_id, decisions=decisions, diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 0764d927b..6a722cccb 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -8,6 +8,14 @@ from cleveragents.domain.models.core.automation_profile import ( GuardResult, get_builtin_profile, ) +from cleveragents.domain.models.core.autonomy_guardrails import ( + ActorLimits, + AutonomyGuardrails, + GuardrailAuditEntry, + GuardrailAuditTrail, + GuardrailEventType, + GuardrailResult, +) from cleveragents.domain.models.core.change import ( Change, ChangeEntry, @@ -224,8 +232,10 @@ __all__ = [ "ROLE_PERMISSIONS", "ActionState", "Actor", + "ActorLimits", "AutomationGuard", "AutomationProfile", + "AutonomyGuardrails", "BindingMode", "BindingResult", "Change", @@ -266,6 +276,10 @@ __all__ = [ "ErrorRecord", "ErrorRecoveryPolicy", "GuardResult", + "GuardrailAuditEntry", + "GuardrailAuditTrail", + "GuardrailEventType", + "GuardrailResult", "InMemoryChangeSetStore", "InMemoryInvocationTracker", "Invariant", diff --git a/src/cleveragents/domain/models/core/autonomy_guardrails.py b/src/cleveragents/domain/models/core/autonomy_guardrails.py new file mode 100644 index 000000000..a7c8c2a4c --- /dev/null +++ b/src/cleveragents/domain/models/core/autonomy_guardrails.py @@ -0,0 +1,458 @@ +"""Autonomy guardrails and audit trail models. + +Extends the automation guard framework with: + +- **AutonomyGuardrails** — runtime constraints for max steps, tool + budget, wall-clock time, per-actor limits, and required confirmations + during plan execution. +- **GuardrailAuditEntry** — a single audit trail entry recording a + guardrail enforcement event with timestamp, event type, result, and + contextual metadata. +- **GuardrailAuditTrail** — a bounded, ordered collection of audit + entries persisted to plan metadata with configurable maximum size + and oldest-first eviction. + +These models support the M6 autonomy hardening milestone by providing +a structured, auditable record of every guardrail decision made during +plan execution. + +Based on ``docs/specification.md`` Section "Automation Profiles" +(autonomy guardrails and audit trail) and "Cost and Rate Limits" +(per-plan budgets including wall-clock time, per-actor limits). +""" + +from __future__ import annotations + +import threading +from datetime import UTC, datetime +from enum import StrEnum + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + + +class GuardrailEventType(StrEnum): + """Types of guardrail enforcement events.""" + + STEP_BLOCKED = "step_blocked" + BUDGET_BLOCKED = "budget_blocked" + CONFIRMATION_REQUIRED = "confirmation_required" + STEP_ALLOWED = "step_allowed" + BUDGET_ALLOWED = "budget_allowed" + CONFIRMATION_GRANTED = "confirmation_granted" + TIME_BLOCKED = "time_blocked" + TIME_ALLOWED = "time_allowed" + ACTOR_LIMIT_BLOCKED = "actor_limit_blocked" + ACTOR_LIMIT_ALLOWED = "actor_limit_allowed" + RETRY_BLOCKED = "retry_blocked" + RETRY_ALLOWED = "retry_allowed" + + +class GuardrailResult(StrEnum): + """Outcome of a guardrail enforcement check.""" + + ALLOWED = "allowed" + DENIED = "denied" + + +class GuardrailAuditEntry(BaseModel): + """A single guardrail enforcement audit trail entry. + + Records the outcome of a guardrail check including the event type, + which guard was evaluated, whether the action was allowed or denied, + and optional contextual metadata. + """ + + timestamp: str = Field( + default_factory=lambda: datetime.now(tz=UTC).isoformat(), + description="ISO-8601 timestamp of the enforcement event", + ) + event_type: GuardrailEventType = Field( + ..., + description="Type of guardrail event", + ) + guard_name: str = Field( + ..., + min_length=1, + description="Name of the guard that was evaluated", + ) + result: GuardrailResult = Field( + ..., + description="Outcome of the guard check (allowed/denied)", + ) + reason: str | None = Field( + default=None, + description="Human-readable reason for the decision", + ) + context: dict[str, str | int | float | bool | None] = Field( + default_factory=dict, + description="Additional context about the enforcement event", + ) + + model_config = ConfigDict( + str_strip_whitespace=True, + ) + + +_DEFAULT_MAX_AUDIT_ENTRIES = 10_000 + + +class GuardrailAuditTrail(BaseModel): + """Bounded, ordered collection of guardrail audit entries. + + Persisted to plan metadata to provide a complete record of + autonomy-related decisions during plan execution. + + When the number of entries exceeds ``max_entries``, the oldest + entries are evicted to keep the trail bounded. + """ + + entries: list[GuardrailAuditEntry] = Field( + default_factory=list, + description="Ordered list of guardrail enforcement entries", + ) + max_entries: int = Field( + default=_DEFAULT_MAX_AUDIT_ENTRIES, + ge=1, + le=1_000_000, + description="Maximum number of audit entries to retain", + ) + _allowed_count: int = 0 + _denied_count: int = 0 + + def model_post_init(self, _context: object) -> None: + """Recompute counters from pre-existing entries after init.""" + self._allowed_count = sum( + 1 for e in self.entries if e.result == GuardrailResult.ALLOWED + ) + self._denied_count = sum( + 1 for e in self.entries if e.result == GuardrailResult.DENIED + ) + + def add_entry(self, entry: GuardrailAuditEntry) -> None: + """Append an audit entry, evicting the oldest if at capacity.""" + if len(self.entries) >= self.max_entries: + evicted = self.entries.pop(0) + if evicted.result == GuardrailResult.ALLOWED: + self._allowed_count -= 1 + else: + self._denied_count -= 1 + self.entries.append(entry) + if entry.result == GuardrailResult.ALLOWED: + self._allowed_count += 1 + else: + self._denied_count += 1 + + @property + def denied_count(self) -> int: + """Count of denied enforcement events.""" + return self._denied_count + + @property + def allowed_count(self) -> int: + """Count of allowed enforcement events.""" + return self._allowed_count + + model_config = ConfigDict( + str_strip_whitespace=True, + ) + + +class ActorLimits(BaseModel): + """Per-actor runtime limits (spec line 27891). + + Controls the maximum number of tool calls per actor invocation and + the maximum number of retries per tool failure within a single actor + step. + """ + + max_tool_calls_per_invocation: int | None = Field( + default=None, + description=( + "Maximum tool calls allowed per actor invocation. None means unlimited." + ), + ) + max_retries_per_failure: int | None = Field( + default=None, + description=( + "Maximum retries per tool failure within an actor step. " + "None means unlimited." + ), + ) + + @field_validator("max_tool_calls_per_invocation") + @classmethod + def validate_max_tool_calls( + cls: type[ActorLimits], + v: int | None, + ) -> int | None: + """Validate max_tool_calls_per_invocation is positive if set.""" + if v is not None and v < 1: + raise ValueError(f"max_tool_calls_per_invocation must be >= 1, got {v}") + return v + + @field_validator("max_retries_per_failure") + @classmethod + def validate_max_retries( + cls: type[ActorLimits], + v: int | None, + ) -> int | None: + """Validate max_retries_per_failure is non-negative if set.""" + if v is not None and v < 0: + raise ValueError(f"max_retries_per_failure must be >= 0, got {v}") + return v + + model_config = ConfigDict( + str_strip_whitespace=True, + ) + + +class AutonomyGuardrails(BaseModel): + """Runtime autonomy constraints for plan execution. + + Extends the automation guard framework with step limits, tool + budgets, wall-clock time limits, per-actor limits, and required + confirmation gates. Tracks current step count and budget spent + for enforcement decisions. + """ + + max_steps: int | None = Field( + default=None, + description=( + "Maximum number of execution steps allowed. None means unlimited." + ), + ) + tool_budget: float | None = Field( + default=None, + description=("Maximum cost budget for tool invocations. None means unlimited."), + ) + max_wall_clock_seconds: float | None = Field( + default=None, + description=( + "Maximum wall-clock time in seconds for plan execution. " + "None means unlimited." + ), + ) + actor_limits: ActorLimits = Field( + default_factory=ActorLimits, + description="Per-actor runtime limits", + ) + required_confirmations: list[str] = Field( + default_factory=list, + description=( + "List of operation names requiring human confirmation before execution." + ), + ) + step_count: int = Field( + default=0, + ge=0, + description="Current execution step counter", + ) + budget_spent: float = Field( + default=0.0, + ge=0.0, + description="Current cumulative budget spent on tool invocations", + ) + start_time: str | None = Field( + default=None, + description="ISO-8601 timestamp of when plan execution started", + ) + + @field_validator("start_time") + @classmethod + def validate_start_time( + cls: type[AutonomyGuardrails], + v: str | None, + ) -> str | None: + """Validate start_time is a parseable ISO-8601 string if set.""" + if v is not None: + try: + datetime.fromisoformat(v) + except (ValueError, TypeError) as exc: + raise ValueError( + f"start_time must be a valid ISO-8601 timestamp, got {v!r}" + ) from exc + return v + + @field_validator("max_steps") + @classmethod + def validate_max_steps( + cls: type[AutonomyGuardrails], + v: int | None, + ) -> int | None: + """Validate max_steps is positive if set.""" + if v is not None and v < 1: + raise ValueError(f"max_steps must be >= 1, got {v}") + return v + + @field_validator("tool_budget") + @classmethod + def validate_tool_budget( + cls: type[AutonomyGuardrails], + v: float | None, + ) -> float | None: + """Validate tool_budget is non-negative if set.""" + if v is not None and v < 0.0: + raise ValueError(f"tool_budget must be >= 0.0, got {v}") + return v + + @field_validator("max_wall_clock_seconds") + @classmethod + def validate_max_wall_clock( + cls: type[AutonomyGuardrails], + v: float | None, + ) -> float | None: + """Validate max_wall_clock_seconds is positive if set.""" + if v is not None and v <= 0.0: + raise ValueError(f"max_wall_clock_seconds must be > 0.0, got {v}") + return v + + @model_validator(mode="after") + def _normalise_confirmations(self) -> AutonomyGuardrails: + """Normalise required_confirmations to lowercase for matching.""" + lowered = [c.lower() for c in self.required_confirmations] + if lowered != self.required_confirmations: + object.__setattr__(self, "required_confirmations", lowered) + return self + + def check_step_limit(self) -> tuple[bool, str | None]: + """Check if the current step count exceeds the max step limit. + + Returns: + A tuple of (allowed, reason). If allowed is False, reason + contains a human-readable explanation. + """ + if self.max_steps is not None and self.step_count >= self.max_steps: + return ( + False, + f"Step limit reached: {self.step_count}/{self.max_steps}", + ) + return (True, None) + + def check_tool_budget(self, tool_cost: float) -> tuple[bool, str | None]: + """Check if the budget allows a tool invocation of the given cost. + + Args: + tool_cost: The cost of the proposed tool invocation. + + Returns: + A tuple of (allowed, reason). + """ + if tool_cost < 0.0: + raise ValueError(f"tool_cost must be >= 0.0, got {tool_cost}") + if self.tool_budget is not None: + projected = self.budget_spent + tool_cost + if projected > self.tool_budget: + return ( + False, + f"Budget exceeded: {projected:.2f} > {self.tool_budget:.2f}", + ) + return (True, None) + + def check_wall_clock(self) -> tuple[bool, str | None]: + """Check if the plan has exceeded its wall-clock time limit. + + Returns: + A tuple of (allowed, reason). + """ + if self.max_wall_clock_seconds is None or self.start_time is None: + return (True, None) + try: + elapsed = ( + datetime.now(UTC) - datetime.fromisoformat(self.start_time) + ).total_seconds() + except (ValueError, TypeError): + return ( + False, + f"Wall-clock check failed: malformed start_time {self.start_time!r}", + ) + if elapsed >= self.max_wall_clock_seconds: + return ( + False, + f"Wall-clock limit reached: {elapsed:.1f}s >= " + f"{self.max_wall_clock_seconds:.1f}s", + ) + return (True, None) + + def check_actor_tool_calls( + self, + current_calls: int, + ) -> tuple[bool, str | None]: + """Check if the actor has exceeded its per-invocation tool call limit. + + Args: + current_calls: Number of tool calls made so far in this invocation. + + Returns: + A tuple of (allowed, reason). + """ + limit = self.actor_limits.max_tool_calls_per_invocation + if limit is not None and current_calls >= limit: + return ( + False, + f"Actor tool-call limit reached: {current_calls}/{limit}", + ) + return (True, None) + + def check_retries_per_failure( + self, + current_retries: int, + ) -> tuple[bool, str | None]: + """Check if the current retry count exceeds the per-failure retry limit. + + Args: + current_retries: Number of retries attempted so far for this failure. + + Returns: + A tuple of (allowed, reason). + """ + limit = self.actor_limits.max_retries_per_failure + if limit is not None and current_retries >= limit: + return ( + False, + f"Retry limit reached: {current_retries}/{limit}", + ) + return (True, None) + + def check_confirmation_required(self, operation: str) -> tuple[bool, str | None]: + """Check if an operation requires human confirmation. + + Args: + operation: The operation name to check (case-insensitive). + + Returns: + A tuple of (confirmation_required, reason). If True, the + operation needs human confirmation before proceeding. + """ + if operation.lower() in self.required_confirmations: + return ( + True, + f"Operation '{operation}' requires human confirmation", + ) + return (False, None) + + def increment_step(self) -> None: + """Increment the step counter by one.""" + self.step_count += 1 + + def record_cost(self, cost: float) -> None: + """Record a tool invocation cost. + + Args: + cost: The cost to add to the budget spent. + """ + if cost < 0.0: + raise ValueError(f"cost must be >= 0.0, got {cost}") + self.budget_spent += cost + + def mark_started(self) -> None: + """Record the plan execution start time for wall-clock tracking.""" + self.start_time = datetime.now(UTC).isoformat() + + model_config = ConfigDict( + str_strip_whitespace=True, + validate_assignment=True, + ) + + +# Re-export threading lock for service-level thread safety +GuardrailLock = threading.Lock diff --git a/vulture_whitelist.py b/vulture_whitelist.py index b52f6e589..4ae2067a2 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -413,3 +413,45 @@ is_local_mode # noqa: B018, F821 get_role_bindings # noqa: B018, F821 add_binding # noqa: B018, F821 remove_binding # noqa: B018, F821 + +# Autonomy guardrails — public API surface +ActorLimits # noqa: B018, F821 +AutonomyGuardrails # noqa: B018, F821 +GuardrailAuditEntry # noqa: B018, F821 +GuardrailAuditTrail # noqa: B018, F821 +GuardrailEventType # noqa: B018, F821 +GuardrailResult # noqa: B018, F821 +GuardrailLock # noqa: B018, F821 +AutonomyGuardrailService # noqa: B018, F821 +autonomy_guardrail_service # noqa: B018, F821 +check_step_limit # noqa: B018, F821 +check_tool_budget # noqa: B018, F821 +check_wall_clock # noqa: B018, F821 +check_actor_tool_calls # noqa: B018, F821 +check_confirmation_required # noqa: B018, F821 +record_enforcement # noqa: B018, F821 +get_audit_trail # noqa: B018, F821 +get_audit_trail_as_metadata # noqa: B018, F821 +load_from_metadata # noqa: B018, F821 +remove_plan # noqa: B018, F821 +configure_guardrails # noqa: B018, F821 +increment_step # noqa: B018, F821 +record_cost # noqa: B018, F821 +mark_started # noqa: B018, F821 +denied_count # noqa: B018, F821 +allowed_count # noqa: B018, F821 +add_entry # noqa: B018, F821 +max_entries # noqa: B018, F821 +max_wall_clock_seconds # noqa: B018, F821 +actor_limits # noqa: B018, F821 +max_tool_calls_per_invocation # noqa: B018, F821 +max_retries_per_failure # noqa: B018, F821 +start_time # noqa: B018, F821 +_DEFAULT_MAX_AUDIT_ENTRIES # noqa: B018, F821 +check_retries_per_failure # noqa: B018, F821 +RETRY_BLOCKED # noqa: B018, F821 +RETRY_ALLOWED # noqa: B018, F821 +validate_start_time # noqa: B018, F821 +_enforce_guardrails # noqa: B018, F821 +_enforce_guardrails_per_step # noqa: B018, F821 +guardrail_service # noqa: B018, F821