feat(automation): add automation profiles and guards
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 21s
CI / lint (pull_request) Successful in 22s
CI / security (pull_request) Successful in 36s
CI / typecheck (pull_request) Successful in 37s
CI / integration_tests (pull_request) Successful in 4m49s
CI / unit_tests (pull_request) Successful in 5m54s
CI / docker (pull_request) Successful in 38s
CI / benchmark-regression (pull_request) Successful in 15m25s
CI / coverage (pull_request) Successful in 41m50s
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 21s
CI / lint (pull_request) Successful in 22s
CI / security (pull_request) Successful in 36s
CI / typecheck (pull_request) Successful in 37s
CI / integration_tests (pull_request) Successful in 4m49s
CI / unit_tests (pull_request) Successful in 5m54s
CI / docker (pull_request) Successful in 38s
CI / benchmark-regression (pull_request) Successful in 15m25s
CI / coverage (pull_request) Successful in 41m50s
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
"""ASV benchmarks for AutomationProfile validation and serialization.
|
||||
"""ASV benchmarks for AutomationProfile validation, serialization, and guards.
|
||||
|
||||
Measures the performance of:
|
||||
- AutomationProfile model construction (Pydantic validation)
|
||||
@@ -6,6 +6,7 @@ Measures the performance of:
|
||||
- AutomationProfile.from_config() factory
|
||||
- Built-in profile lookup via get_builtin_profile()
|
||||
- BUILTIN_PROFILES iteration
|
||||
- AutomationGuard check_guard() enforcement overhead
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -14,6 +15,9 @@ import sys
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationProfile,
|
||||
@@ -21,6 +25,9 @@ try:
|
||||
)
|
||||
except ModuleNotFoundError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationProfile,
|
||||
@@ -124,3 +131,61 @@ class BuiltinProfileSuite:
|
||||
"""Benchmark iterating all built-in profiles."""
|
||||
for name in BUILTIN_PROFILES:
|
||||
_ = BUILTIN_PROFILES[name]
|
||||
|
||||
|
||||
class GuardCheckSuite:
|
||||
"""Benchmark AutomationGuard check_guard() operations."""
|
||||
|
||||
def setup(self) -> None:
|
||||
"""Create profiles with various guard configurations."""
|
||||
self.no_guard_profile = AutomationProfile(name="bench/no-guard")
|
||||
self.guarded_profile = AutomationProfile(
|
||||
name="bench/guarded",
|
||||
guards=AutomationGuard(
|
||||
max_tool_calls_per_step=10,
|
||||
max_total_cost=100.0,
|
||||
tool_denylist=["dangerous_tool"],
|
||||
tool_allowlist=["safe_tool", "read_file", "list_dir"],
|
||||
require_approval_for_writes=True,
|
||||
require_approval_for_apply=True,
|
||||
),
|
||||
)
|
||||
self.minimal_guard = AutomationProfile(
|
||||
name="bench/minimal-guard",
|
||||
guards=AutomationGuard(max_tool_calls_per_step=50),
|
||||
)
|
||||
|
||||
def time_check_guard_no_guards(self) -> None:
|
||||
"""Benchmark check_guard with no guards configured."""
|
||||
self.no_guard_profile.check_guard("any_tool")
|
||||
|
||||
def time_check_guard_allowed(self) -> None:
|
||||
"""Benchmark check_guard when tool is allowed."""
|
||||
self.guarded_profile.check_guard("safe_tool", calls_so_far=5)
|
||||
|
||||
def time_check_guard_denied(self) -> None:
|
||||
"""Benchmark check_guard when tool is on denylist."""
|
||||
self.guarded_profile.check_guard("dangerous_tool")
|
||||
|
||||
def time_check_guard_write(self) -> None:
|
||||
"""Benchmark check_guard with write flag."""
|
||||
self.guarded_profile.check_guard("safe_tool", is_write=True)
|
||||
|
||||
def time_check_guard_cost(self) -> None:
|
||||
"""Benchmark check_guard with cost check."""
|
||||
self.guarded_profile.check_guard(
|
||||
"safe_tool",
|
||||
cost_so_far=50.0,
|
||||
)
|
||||
|
||||
def time_check_guard_minimal(self) -> None:
|
||||
"""Benchmark check_guard with minimal guard config."""
|
||||
self.minimal_guard.check_guard("tool", calls_so_far=10)
|
||||
|
||||
def time_guard_construction(self) -> None:
|
||||
"""Benchmark AutomationGuard model construction."""
|
||||
AutomationGuard(
|
||||
max_tool_calls_per_step=10,
|
||||
max_total_cost=100.0,
|
||||
tool_denylist=["tool_a", "tool_b"],
|
||||
)
|
||||
|
||||
@@ -109,6 +109,62 @@ allow_unsafe_tools: false
|
||||
|
||||
See `docs/schema/automation_profile.schema.yaml` for the full YAML schema and `examples/profiles/` for example configurations.
|
||||
|
||||
## Automation Guards
|
||||
|
||||
Guards are optional enforcement hooks attached to a profile that gate tool invocations at runtime. They operate independently of the phase-transition thresholds and provide fine-grained control over tool usage.
|
||||
|
||||
### Guard Fields
|
||||
|
||||
| Field | Type | Default | Description |
|
||||
|-------|------|---------|-------------|
|
||||
| `max_tool_calls_per_step` | `int \| None` | `None` | Maximum tool invocations per step before requiring approval |
|
||||
| `max_total_cost` | `float \| None` | `None` | Budget cap before requiring approval |
|
||||
| `tool_allowlist` | `list[str] \| None` | `None` | Only these tools can be called automatically |
|
||||
| `tool_denylist` | `list[str] \| None` | `None` | These tools always require approval |
|
||||
| `require_approval_for_writes` | `bool` | `false` | Require human approval for write operations |
|
||||
| `require_approval_for_apply` | `bool` | `false` | Require human approval before apply phase |
|
||||
|
||||
### Guard Evaluation Order
|
||||
|
||||
When `check_guard()` is called, guards are evaluated in this order:
|
||||
|
||||
1. **Denylist** — If the tool is on the denylist, it is blocked immediately.
|
||||
2. **Allowlist** — If an allowlist is set and the tool is not on it, it is blocked.
|
||||
3. **Max tool calls** — If the call count has reached the limit, approval is required.
|
||||
4. **Cost budget** — If the cumulative cost has reached the cap, approval is required.
|
||||
5. **Write approval** — If the operation is a write and write approval is required.
|
||||
6. **Apply approval** — If the tool name is `__apply__` and apply approval is required.
|
||||
|
||||
If no guard blocks the invocation, it is allowed.
|
||||
|
||||
### GuardResult
|
||||
|
||||
The result of a guard evaluation is a `GuardResult` with three fields:
|
||||
|
||||
- `allowed` (`bool`) — Whether the tool invocation may proceed.
|
||||
- `reason` (`str | None`) — Human-readable explanation when blocked.
|
||||
- `requires_approval` (`bool`) — Whether human approval can unblock the invocation.
|
||||
|
||||
### Example YAML with Guards
|
||||
|
||||
```yaml
|
||||
name: acme/guarded
|
||||
description: Profile with guard constraints
|
||||
schema_version: "1.0"
|
||||
auto_strategize: 0.7
|
||||
auto_execute: 0.7
|
||||
auto_apply: 1.0
|
||||
|
||||
guards:
|
||||
max_tool_calls_per_step: 10
|
||||
max_total_cost: 5.0
|
||||
tool_denylist:
|
||||
- shell_exec
|
||||
- file_delete
|
||||
require_approval_for_writes: true
|
||||
require_approval_for_apply: true
|
||||
```
|
||||
|
||||
## CLI Commands
|
||||
|
||||
The `agents automation-profile` command group manages automation profiles
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
# Custom profile: custom-cautious
|
||||
# Based on the built-in 'cautious' profile with additional guard
|
||||
# constraints for controlled environments.
|
||||
|
||||
name: custom/cautious
|
||||
description: Cautious profile with guard constraints
|
||||
schema_version: "1.0"
|
||||
|
||||
# Phase-transition thresholds
|
||||
auto_strategize: 0.7
|
||||
auto_execute: 0.7
|
||||
auto_apply: 1.0
|
||||
|
||||
# Decision-autonomy thresholds
|
||||
auto_decisions_strategize: 0.6
|
||||
auto_decisions_execute: 0.8
|
||||
|
||||
# Self-repair thresholds
|
||||
auto_validation_fix: 0.7
|
||||
auto_strategy_revision: 0.8
|
||||
auto_reversion_from_apply: 0.9
|
||||
|
||||
# Child plan and retry thresholds
|
||||
auto_child_plans: 0.7
|
||||
auto_retry_transient: 0.0
|
||||
auto_checkpoint_restore: 0.6
|
||||
|
||||
# Safety requirements
|
||||
require_sandbox: true
|
||||
require_checkpoints: true
|
||||
allow_unsafe_tools: false
|
||||
|
||||
# Guard constraints
|
||||
guards:
|
||||
max_tool_calls_per_step: 10
|
||||
max_total_cost: 5.0
|
||||
tool_denylist:
|
||||
- shell_exec
|
||||
- file_delete
|
||||
require_approval_for_writes: true
|
||||
require_approval_for_apply: true
|
||||
@@ -0,0 +1,214 @@
|
||||
Feature: Automation Profile Guards
|
||||
As a developer
|
||||
I want automation profile guards to enforce runtime constraints
|
||||
So that tool invocations are gated by budget, call limits, and policies
|
||||
|
||||
# ---- Guard allows tool within limits ----
|
||||
|
||||
Scenario: Guard allows tool within call limits
|
||||
Given a profile with max_tool_calls_per_step 5
|
||||
When I check guard for tool "read_file" with 3 calls so far
|
||||
Then the guard result should be allowed
|
||||
|
||||
Scenario: Guard allows tool with no guards configured
|
||||
Given a profile with no guards
|
||||
When I check guard for tool "any_tool"
|
||||
Then the guard result should be allowed
|
||||
|
||||
# ---- Guard blocks tool exceeding max_tool_calls ----
|
||||
|
||||
Scenario: Guard blocks tool exceeding max_tool_calls
|
||||
Given a profile with max_tool_calls_per_step 3
|
||||
When I check guard for tool "read_file" with 3 calls so far
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "limit"
|
||||
And the guard should require approval
|
||||
|
||||
Scenario: Guard blocks tool at exact max_tool_calls boundary
|
||||
Given a profile with max_tool_calls_per_step 0
|
||||
When I check guard for tool "read_file" with 0 calls so far
|
||||
Then the guard result should not be allowed
|
||||
|
||||
# ---- Guard blocks tool on denylist ----
|
||||
|
||||
Scenario: Guard blocks tool on denylist
|
||||
Given a profile with denylist "shell_exec,file_delete"
|
||||
When I check guard for tool "shell_exec"
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "denylist"
|
||||
And the guard should require approval
|
||||
|
||||
Scenario: Guard allows tool not on denylist
|
||||
Given a profile with denylist "shell_exec,file_delete"
|
||||
When I check guard for tool "read_file"
|
||||
Then the guard result should be allowed
|
||||
|
||||
# ---- Guard allows tool on allowlist ----
|
||||
|
||||
Scenario: Guard allows tool on allowlist
|
||||
Given a profile with allowlist "read_file,list_dir"
|
||||
When I check guard for tool "read_file"
|
||||
Then the guard result should be allowed
|
||||
|
||||
Scenario: Guard blocks tool not on allowlist
|
||||
Given a profile with allowlist "read_file,list_dir"
|
||||
When I check guard for tool "write_file"
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "allowlist"
|
||||
|
||||
# ---- Guard requires approval for writes ----
|
||||
|
||||
Scenario: Guard requires approval for write operations
|
||||
Given a profile with require_approval_for_writes true
|
||||
When I check guard for tool "write_file" with is_write true
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "Write"
|
||||
And the guard should require approval
|
||||
|
||||
Scenario: Guard allows non-write operations when writes require approval
|
||||
Given a profile with require_approval_for_writes true
|
||||
When I check guard for tool "read_file" with is_write false
|
||||
Then the guard result should be allowed
|
||||
|
||||
# ---- Guard requires approval for apply ----
|
||||
|
||||
Scenario: Guard requires approval for apply phase
|
||||
Given a profile with require_approval_for_apply true
|
||||
When I check guard for tool "__apply__"
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "Apply"
|
||||
And the guard should require approval
|
||||
|
||||
Scenario: Guard allows non-apply tools when apply requires approval
|
||||
Given a profile with require_approval_for_apply true
|
||||
When I check guard for tool "read_file"
|
||||
Then the guard result should be allowed
|
||||
|
||||
# ---- Cost budget enforcement ----
|
||||
|
||||
Scenario: Guard blocks tool when cost budget exceeded
|
||||
Given a profile with max_total_cost 10.0
|
||||
When I check guard for tool "expensive_tool" with cost 10.0
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "Cost"
|
||||
And the guard should require approval
|
||||
|
||||
Scenario: Guard allows tool within cost budget
|
||||
Given a profile with max_total_cost 10.0
|
||||
When I check guard for tool "cheap_tool" with cost 5.0
|
||||
Then the guard result should be allowed
|
||||
|
||||
# ---- Effective profile resolution ----
|
||||
|
||||
Scenario: Effective profile resolves plan-level first
|
||||
Given a profile service with profiles "cautious" and "manual"
|
||||
When I resolve effective profile with plan "cautious"
|
||||
Then the resolved guard profile name should be "cautious"
|
||||
|
||||
Scenario: Effective profile falls back to action level
|
||||
Given a profile service with profiles "cautious" and "manual"
|
||||
When I resolve effective profile with action "cautious"
|
||||
Then the resolved guard profile name should be "cautious"
|
||||
|
||||
Scenario: Effective profile falls back to default
|
||||
Given a profile service with profiles "cautious" and "manual"
|
||||
When I resolve effective profile with default "cautious"
|
||||
Then the resolved guard profile name should be "cautious"
|
||||
|
||||
Scenario: Effective profile uses global when no overrides
|
||||
Given a profile service with global default "manual"
|
||||
When I resolve effective profile with no overrides
|
||||
Then the resolved guard profile name should be "manual"
|
||||
|
||||
# ---- Custom profile YAML loading ----
|
||||
|
||||
Scenario: Load custom profile from YAML file
|
||||
Given a YAML profile file with guards
|
||||
When I load the profile from YAML
|
||||
Then the loaded profile should have guards
|
||||
And the loaded profile guards max_tool_calls_per_step should be 10
|
||||
|
||||
Scenario: Load custom profile from YAML via service
|
||||
Given a YAML profile file with guards
|
||||
When I load the profile via service from_yaml
|
||||
Then the loaded profile should have guards
|
||||
|
||||
# ---- Built-in profile immutability ----
|
||||
|
||||
Scenario: Built-in profiles cannot be updated
|
||||
Given a profile service
|
||||
When I try to update built-in profile "manual"
|
||||
Then a profile service validation error should be raised
|
||||
|
||||
Scenario: Built-in profiles cannot be deleted
|
||||
Given a profile service
|
||||
When I try to delete built-in profile "manual"
|
||||
Then a profile service validation error should be raised
|
||||
|
||||
# ---- Guard model validation ----
|
||||
|
||||
Scenario: AutomationGuard rejects negative max_tool_calls
|
||||
When I try to create a guard with max_tool_calls -1
|
||||
Then a guard model validation error should be raised
|
||||
|
||||
Scenario: AutomationGuard rejects negative max_total_cost
|
||||
When I try to create a guard with max_total_cost -1.0
|
||||
Then a guard model validation error should be raised
|
||||
|
||||
Scenario: AutomationGuard accepts zero max_tool_calls
|
||||
When I create a guard with max_tool_calls 0
|
||||
Then the guard should be created
|
||||
|
||||
Scenario: AutomationGuard accepts None values
|
||||
When I create a guard with defaults
|
||||
Then the guard max_tool_calls_per_step should be None
|
||||
And the guard max_total_cost should be None
|
||||
|
||||
# ---- GuardResult model ----
|
||||
|
||||
Scenario: GuardResult allowed with no reason
|
||||
When I create a guard result allowed true
|
||||
Then the guard result allowed should be true
|
||||
And the guard result reason should be None
|
||||
|
||||
Scenario: GuardResult denied with reason
|
||||
When I create a guard result denied with reason "blocked"
|
||||
Then the guard result allowed should be false
|
||||
And the guard result reason should be "blocked"
|
||||
And the guard result requires_approval should be true
|
||||
|
||||
# ---- Service evaluate_guard ----
|
||||
|
||||
Scenario: Service evaluate_guard delegates to profile
|
||||
Given a profile service
|
||||
When I evaluate guard for profile "full-auto" tool "any_tool"
|
||||
Then the guard result should be allowed
|
||||
|
||||
Scenario: Service evaluate_guard rejects empty profile name
|
||||
Given a profile service
|
||||
When I try to evaluate guard with empty profile name
|
||||
Then a profile service validation error should be raised
|
||||
|
||||
Scenario: Service evaluate_guard rejects empty tool name
|
||||
Given a profile service
|
||||
When I try to evaluate guard with empty tool name
|
||||
Then a profile service validation error should be raised
|
||||
|
||||
# ---- Profile with guards from config dict ----
|
||||
|
||||
Scenario: Profile from config dict with guards
|
||||
When I create a profile from config with guards
|
||||
Then the profile should have guards configured
|
||||
And the profile guards should have denylist
|
||||
|
||||
# ---- check_guard argument validation ----
|
||||
|
||||
Scenario: check_guard rejects negative cost_so_far
|
||||
Given a profile with max_total_cost 10.0
|
||||
When I try to check guard with negative cost
|
||||
Then a guard argument error should be raised
|
||||
|
||||
Scenario: check_guard rejects negative calls_so_far
|
||||
Given a profile with max_tool_calls_per_step 5
|
||||
When I try to check guard with negative calls
|
||||
Then a guard argument error should be raised
|
||||
@@ -0,0 +1,475 @@
|
||||
"""Step definitions for Automation Profile Guards tests."""
|
||||
|
||||
import tempfile
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
from pydantic import ValidationError
|
||||
|
||||
from cleveragents.application.services.automation_profile_service import (
|
||||
AutomationProfileService,
|
||||
)
|
||||
from cleveragents.core.exceptions import ValidationError as CAValidationError
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
GuardResult,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
AutomationProfile,
|
||||
)
|
||||
|
||||
|
||||
def _make_guarded_profile(**guard_kwargs: Any) -> AutomationProfile:
|
||||
"""Create a profile with specific guard settings."""
|
||||
guards = AutomationGuard(**guard_kwargs)
|
||||
return AutomationProfile(name="test/guarded", guards=guards)
|
||||
|
||||
|
||||
# -- Guard setup steps --
|
||||
|
||||
|
||||
@given("a profile with max_tool_calls_per_step {limit:d}")
|
||||
def step_profile_max_calls(context: Context, limit: int) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
max_tool_calls_per_step=limit,
|
||||
)
|
||||
|
||||
|
||||
@given("a profile with no guards")
|
||||
def step_profile_no_guards(context: Context) -> None:
|
||||
context.guard_profile = AutomationProfile(name="test/no-guards")
|
||||
|
||||
|
||||
@given('a profile with denylist "{tools}"')
|
||||
def step_profile_denylist(context: Context, tools: str) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
tool_denylist=tools.split(","),
|
||||
)
|
||||
|
||||
|
||||
@given('a profile with allowlist "{tools}"')
|
||||
def step_profile_allowlist(context: Context, tools: str) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
tool_allowlist=tools.split(","),
|
||||
)
|
||||
|
||||
|
||||
@given("a profile with require_approval_for_writes true")
|
||||
def step_profile_write_approval(context: Context) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
require_approval_for_writes=True,
|
||||
)
|
||||
|
||||
|
||||
@given("a profile with require_approval_for_apply true")
|
||||
def step_profile_apply_approval(context: Context) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
require_approval_for_apply=True,
|
||||
)
|
||||
|
||||
|
||||
@given("a profile with max_total_cost {cost:g}")
|
||||
def step_profile_max_cost(context: Context, cost: float) -> None:
|
||||
context.guard_profile = _make_guarded_profile(
|
||||
max_total_cost=cost,
|
||||
)
|
||||
|
||||
|
||||
# -- Guard check steps --
|
||||
|
||||
|
||||
@when('I check guard for tool "{tool}" with {calls:d} calls so far')
|
||||
def step_check_guard_calls(
|
||||
context: Context,
|
||||
tool: str,
|
||||
calls: int,
|
||||
) -> None:
|
||||
context.guard_result = context.guard_profile.check_guard(
|
||||
tool_name=tool,
|
||||
calls_so_far=calls,
|
||||
)
|
||||
|
||||
|
||||
@when('I check guard for tool "{tool}"')
|
||||
def step_check_guard_simple(context: Context, tool: str) -> None:
|
||||
context.guard_result = context.guard_profile.check_guard(
|
||||
tool_name=tool,
|
||||
)
|
||||
|
||||
|
||||
@when('I check guard for tool "{tool}" with is_write {is_w}')
|
||||
def step_check_guard_write(
|
||||
context: Context,
|
||||
tool: str,
|
||||
is_w: str,
|
||||
) -> None:
|
||||
context.guard_result = context.guard_profile.check_guard(
|
||||
tool_name=tool,
|
||||
is_write=is_w.lower() == "true",
|
||||
)
|
||||
|
||||
|
||||
@when('I check guard for tool "{tool}" with cost {cost:g}')
|
||||
def step_check_guard_cost(
|
||||
context: Context,
|
||||
tool: str,
|
||||
cost: float,
|
||||
) -> None:
|
||||
context.guard_result = context.guard_profile.check_guard(
|
||||
tool_name=tool,
|
||||
cost_so_far=cost,
|
||||
)
|
||||
|
||||
|
||||
# -- Guard result assertions --
|
||||
|
||||
|
||||
@then("the guard result should be allowed")
|
||||
def step_guard_allowed(context: Context) -> None:
|
||||
assert context.guard_result.allowed is True, (
|
||||
f"Expected allowed, got: {context.guard_result}"
|
||||
)
|
||||
|
||||
|
||||
@then("the guard result should not be allowed")
|
||||
def step_guard_not_allowed(context: Context) -> None:
|
||||
assert context.guard_result.allowed is False, (
|
||||
f"Expected not allowed, got: {context.guard_result}"
|
||||
)
|
||||
|
||||
|
||||
@then('the guard reason should mention "{text}"')
|
||||
def step_guard_reason_mention(context: Context, text: str) -> None:
|
||||
reason = context.guard_result.reason or ""
|
||||
assert text in reason, f"Expected reason to mention '{text}', got: '{reason}'"
|
||||
|
||||
|
||||
@then("the guard should require approval")
|
||||
def step_guard_requires_approval(context: Context) -> None:
|
||||
assert context.guard_result.requires_approval is True
|
||||
|
||||
|
||||
# -- Effective profile resolution steps --
|
||||
|
||||
|
||||
@given('a profile service with profiles "{p1}" and "{p2}"')
|
||||
def step_service_with_profiles(
|
||||
context: Context,
|
||||
p1: str,
|
||||
p2: str,
|
||||
) -> None:
|
||||
context.profile_service = AutomationProfileService()
|
||||
|
||||
|
||||
@given('a profile service with global default "{name}"')
|
||||
def step_service_global_default(context: Context, name: str) -> None:
|
||||
context.profile_service = AutomationProfileService(
|
||||
global_default=name,
|
||||
)
|
||||
|
||||
|
||||
@given("a profile service")
|
||||
def step_service_default(context: Context) -> None:
|
||||
context.profile_service = AutomationProfileService()
|
||||
|
||||
|
||||
@when('I resolve effective profile with plan "{name}"')
|
||||
def step_resolve_plan(context: Context, name: str) -> None:
|
||||
context.resolved_profile = context.profile_service.get_effective_profile(
|
||||
plan_profile=name,
|
||||
)
|
||||
|
||||
|
||||
@when('I resolve effective profile with action "{name}"')
|
||||
def step_resolve_action(context: Context, name: str) -> None:
|
||||
context.resolved_profile = context.profile_service.get_effective_profile(
|
||||
action_profile=name,
|
||||
)
|
||||
|
||||
|
||||
@when('I resolve effective profile with default "{name}"')
|
||||
def step_resolve_default(context: Context, name: str) -> None:
|
||||
context.resolved_profile = context.profile_service.get_effective_profile(
|
||||
default_profile=name,
|
||||
)
|
||||
|
||||
|
||||
@when("I resolve effective profile with no overrides")
|
||||
def step_resolve_no_overrides(context: Context) -> None:
|
||||
context.resolved_profile = context.profile_service.get_effective_profile()
|
||||
|
||||
|
||||
@then('the resolved guard profile name should be "{name}"')
|
||||
def step_resolved_guard_name(context: Context, name: str) -> None:
|
||||
assert context.resolved_profile.name == name, (
|
||||
f"Expected '{name}', got '{context.resolved_profile.name}'"
|
||||
)
|
||||
|
||||
|
||||
# -- YAML loading steps --
|
||||
|
||||
|
||||
@given("a YAML profile file with guards")
|
||||
def step_yaml_file_with_guards(context: Context) -> None:
|
||||
config = {
|
||||
"name": "test/yaml-guards",
|
||||
"description": "YAML test profile",
|
||||
"guards": {
|
||||
"max_tool_calls_per_step": 10,
|
||||
"max_total_cost": 50.0,
|
||||
"require_approval_for_writes": True,
|
||||
},
|
||||
}
|
||||
tmp_path = tempfile.mktemp(suffix=".yaml")
|
||||
with open(tmp_path, "w") as fh:
|
||||
yaml.dump(config, fh)
|
||||
context.yaml_path = tmp_path
|
||||
|
||||
|
||||
@when("I load the profile from YAML")
|
||||
def step_load_yaml(context: Context) -> None:
|
||||
context.loaded_profile = AutomationProfile.from_yaml(
|
||||
context.yaml_path,
|
||||
)
|
||||
|
||||
|
||||
@when("I load the profile via service from_yaml")
|
||||
def step_load_yaml_service(context: Context) -> None:
|
||||
context.loaded_profile = AutomationProfileService.from_yaml(
|
||||
context.yaml_path,
|
||||
)
|
||||
|
||||
|
||||
@then("the loaded profile should have guards")
|
||||
def step_loaded_has_guards(context: Context) -> None:
|
||||
assert context.loaded_profile.guards is not None
|
||||
|
||||
|
||||
@then("the loaded profile guards max_tool_calls_per_step should be {val:d}")
|
||||
def step_loaded_max_calls(context: Context, val: int) -> None:
|
||||
assert context.loaded_profile.guards is not None
|
||||
actual = context.loaded_profile.guards.max_tool_calls_per_step
|
||||
assert actual == val, f"Expected {val}, got {actual}"
|
||||
|
||||
|
||||
# -- Built-in immutability steps --
|
||||
|
||||
|
||||
@when('I try to update built-in profile "{name}"')
|
||||
def step_try_update_builtin(context: Context, name: str) -> None:
|
||||
context.service_error = None
|
||||
try:
|
||||
context.profile_service.update_profile(name, {"name": name})
|
||||
except CAValidationError as e:
|
||||
context.service_error = e
|
||||
|
||||
|
||||
@when('I try to delete built-in profile "{name}"')
|
||||
def step_try_delete_builtin(context: Context, name: str) -> None:
|
||||
context.service_error = None
|
||||
try:
|
||||
context.profile_service.delete_profile(name)
|
||||
except CAValidationError as e:
|
||||
context.service_error = e
|
||||
|
||||
|
||||
@then("a profile service validation error should be raised")
|
||||
def step_service_validation_error(context: Context) -> None:
|
||||
assert context.service_error is not None, "Expected a ValidationError"
|
||||
|
||||
|
||||
# -- Guard model validation steps --
|
||||
|
||||
|
||||
@when("I try to create a guard with max_tool_calls {val:d}")
|
||||
def step_try_guard_max_calls(context: Context, val: int) -> None:
|
||||
context.guard_error = None
|
||||
context.guard_model = None
|
||||
try:
|
||||
context.guard_model = AutomationGuard(
|
||||
max_tool_calls_per_step=val,
|
||||
)
|
||||
except ValidationError as e:
|
||||
context.guard_error = e
|
||||
|
||||
|
||||
@when("I try to create a guard with max_total_cost {val:g}")
|
||||
def step_try_guard_max_cost(context: Context, val: float) -> None:
|
||||
context.guard_error = None
|
||||
context.guard_model = None
|
||||
try:
|
||||
context.guard_model = AutomationGuard(max_total_cost=val)
|
||||
except ValidationError as e:
|
||||
context.guard_error = e
|
||||
|
||||
|
||||
@when("I create a guard with max_tool_calls {val:d}")
|
||||
def step_create_guard_max_calls(context: Context, val: int) -> None:
|
||||
context.guard_model = AutomationGuard(
|
||||
max_tool_calls_per_step=val,
|
||||
)
|
||||
|
||||
|
||||
@when("I create a guard with defaults")
|
||||
def step_create_guard_defaults(context: Context) -> None:
|
||||
context.guard_model = AutomationGuard()
|
||||
|
||||
|
||||
@then("a guard model validation error should be raised")
|
||||
def step_guard_model_validation_error(context: Context) -> None:
|
||||
assert context.guard_error is not None, "Expected a validation error"
|
||||
|
||||
|
||||
@then("the guard result allowed should be true")
|
||||
def step_result_allowed_true(context: Context) -> None:
|
||||
assert context.guard_result.allowed is True
|
||||
|
||||
|
||||
@then("the guard result allowed should be false")
|
||||
def step_result_allowed_false(context: Context) -> None:
|
||||
assert context.guard_result.allowed is False
|
||||
|
||||
|
||||
@then("the guard result reason should be None")
|
||||
def step_result_reason_none(context: Context) -> None:
|
||||
assert context.guard_result.reason is None
|
||||
|
||||
|
||||
@then('the guard result reason should be "{expected}"')
|
||||
def step_result_reason_value(context: Context, expected: str) -> None:
|
||||
assert context.guard_result.reason == expected
|
||||
|
||||
|
||||
@then("the guard result requires_approval should be true")
|
||||
def step_result_requires_approval(context: Context) -> None:
|
||||
assert context.guard_result.requires_approval is True
|
||||
|
||||
|
||||
# -- Service evaluate_guard steps --
|
||||
|
||||
|
||||
@when('I evaluate guard for profile "{name}" tool "{tool}"')
|
||||
def step_evaluate_guard(context: Context, name: str, tool: str) -> None:
|
||||
context.guard_result = context.profile_service.evaluate_guard(
|
||||
profile_name=name,
|
||||
tool_name=tool,
|
||||
)
|
||||
|
||||
|
||||
@when("I try to evaluate guard with empty profile name")
|
||||
def step_evaluate_empty_profile(context: Context) -> None:
|
||||
context.service_error = None
|
||||
try:
|
||||
context.profile_service.evaluate_guard(
|
||||
profile_name="",
|
||||
tool_name="tool",
|
||||
)
|
||||
except CAValidationError as e:
|
||||
context.service_error = e
|
||||
|
||||
|
||||
@when("I try to evaluate guard with empty tool name")
|
||||
def step_evaluate_empty_tool(context: Context) -> None:
|
||||
context.service_error = None
|
||||
try:
|
||||
context.profile_service.evaluate_guard(
|
||||
profile_name="manual",
|
||||
tool_name="",
|
||||
)
|
||||
except CAValidationError as e:
|
||||
context.service_error = e
|
||||
|
||||
|
||||
# -- Profile with guards from config --
|
||||
|
||||
|
||||
@when("I create a profile from config with guards")
|
||||
def step_profile_config_guards(context: Context) -> None:
|
||||
config: dict[str, Any] = {
|
||||
"name": "test/config-guards",
|
||||
"guards": {
|
||||
"tool_denylist": ["dangerous"],
|
||||
"require_approval_for_writes": True,
|
||||
},
|
||||
}
|
||||
context.config_profile = AutomationProfile.from_config(config)
|
||||
|
||||
|
||||
@then("the profile should have guards configured")
|
||||
def step_profile_has_guards(context: Context) -> None:
|
||||
assert context.config_profile.guards is not None
|
||||
|
||||
|
||||
@then("the profile guards should have denylist")
|
||||
def step_profile_guards_denylist(context: Context) -> None:
|
||||
assert context.config_profile.guards is not None
|
||||
assert context.config_profile.guards.tool_denylist is not None
|
||||
assert len(context.config_profile.guards.tool_denylist) > 0
|
||||
|
||||
|
||||
# -- check_guard argument validation --
|
||||
|
||||
|
||||
@when("I try to check guard with negative cost")
|
||||
def step_check_guard_negative_cost(context: Context) -> None:
|
||||
context.guard_arg_error = None
|
||||
try:
|
||||
context.guard_profile.check_guard(
|
||||
tool_name="tool",
|
||||
cost_so_far=-1.0,
|
||||
)
|
||||
except ValueError as e:
|
||||
context.guard_arg_error = e
|
||||
|
||||
|
||||
@when("I try to check guard with negative calls")
|
||||
def step_check_guard_negative_calls(context: Context) -> None:
|
||||
context.guard_arg_error = None
|
||||
try:
|
||||
context.guard_profile.check_guard(
|
||||
tool_name="tool",
|
||||
calls_so_far=-1,
|
||||
)
|
||||
except ValueError as e:
|
||||
context.guard_arg_error = e
|
||||
|
||||
|
||||
@then("a guard argument error should be raised")
|
||||
def step_guard_arg_error(context: Context) -> None:
|
||||
assert context.guard_arg_error is not None, (
|
||||
"Expected a ValueError for invalid arguments"
|
||||
)
|
||||
|
||||
|
||||
# -- Missing model construction / assertion steps --
|
||||
|
||||
|
||||
@then("the guard should be created")
|
||||
def step_guard_should_be_created(context: Context) -> None:
|
||||
assert context.guard_model is not None, "Guard was not created"
|
||||
|
||||
|
||||
@then("the guard max_tool_calls_per_step should be None")
|
||||
def step_guard_max_calls_none(context: Context) -> None:
|
||||
assert context.guard_model.max_tool_calls_per_step is None
|
||||
|
||||
|
||||
@then("the guard max_total_cost should be None")
|
||||
def step_guard_max_cost_none(context: Context) -> None:
|
||||
assert context.guard_model.max_total_cost is None
|
||||
|
||||
|
||||
@when("I create a guard result allowed true")
|
||||
def step_create_guard_result_allowed(context: Context) -> None:
|
||||
context.guard_result = GuardResult(allowed=True)
|
||||
|
||||
|
||||
@when('I create a guard result denied with reason "{reason}"')
|
||||
def step_create_guard_result_denied(context: Context, reason: str) -> None:
|
||||
context.guard_result = GuardResult(
|
||||
allowed=False,
|
||||
reason=reason,
|
||||
requires_approval=True,
|
||||
)
|
||||
+16
-16
@@ -2692,22 +2692,22 @@ This section replaces all unfinished work items with the Day 14 plan. Work is or
|
||||
- [ ] Git [Hamza]: `git push -u origin feature/m5-context-indexing`
|
||||
- [ ] Forgejo PR [Hamza]: Open PR from `feature/m5-context-indexing` to `master` with a suitable and thorough description
|
||||
|
||||
- [ ] **COMMIT (Owner: Jeff | Group: M5.3.automation-profiles | Branch: feature/m5-automation-profiles | Planned: Day 27 | Expected: Day 28) - Commit message: "feat(automation): add automation profiles and guards"**
|
||||
- [ ] Git [Jeff]: `git checkout master`
|
||||
- [ ] Git [Jeff]: `git pull origin master`
|
||||
- [ ] Git [Jeff]: `git checkout -b feature/m5-automation-profiles`
|
||||
- [ ] Code [Jeff]: Implement automation profiles and enforcement hooks (approval gates, max tool calls, allowlist/denylist).
|
||||
- [ ] Docs [Jeff]: Update `docs/reference/automation_profiles.md` with profile defaults and CLI usage.
|
||||
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
|
||||
- [ ] Tests (Behave) [Jeff]: Add scenarios for profile enforcement and override behavior.
|
||||
- [ ] Tests (Robot) [Jeff]: Add Robot tests for automation profile CLI flags.
|
||||
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/automation_profiles_bench.py` for guard overhead.
|
||||
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
|
||||
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
|
||||
- [ ] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
|
||||
- [ ] Git [Jeff]: `git commit -m "feat(automation): add automation profiles and guards"`
|
||||
- [ ] Git [Jeff]: `git push -u origin feature/m5-automation-profiles`
|
||||
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m5-automation-profiles` to `master` with a suitable and thorough description
|
||||
- [X] **COMMIT (Owner: Jeff | Group: M5.3.automation-profiles | Branch: feature/m5-automation-profiles | Planned: Day 27 | Expected: Day 28) - Commit message: "feat(automation): add automation profiles and guards"** Done: 2026-02-22
|
||||
- [X] Git [Jeff]: `git checkout master`
|
||||
- [X] Git [Jeff]: `git pull origin master`
|
||||
- [X] Git [Jeff]: `git checkout -b feature/m5-automation-profiles`
|
||||
- [X] Code [Jeff]: Implement automation profiles and enforcement hooks (approval gates, max tool calls, allowlist/denylist).
|
||||
- [X] Docs [Jeff]: Update `docs/reference/automation_profiles.md` with profile defaults and CLI usage.
|
||||
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
|
||||
- [X] Tests (Behave) [Jeff]: Add scenarios for profile enforcement and override behavior.
|
||||
- [X] Tests (Robot) [Jeff]: Add Robot tests for automation profile CLI flags.
|
||||
- [X] Tests (ASV) [Jeff]: Add `benchmarks/automation_profiles_bench.py` for guard overhead.
|
||||
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
|
||||
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
|
||||
- [X] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
|
||||
- [X] Git [Jeff]: `git commit -m "feat(automation): add automation profiles and guards"`
|
||||
- [X] Git [Jeff]: `git push -u origin feature/m5-automation-profiles`
|
||||
- [X] Forgejo PR [Jeff]: Open PR from `feature/m5-automation-profiles` to `master` with a suitable and thorough description
|
||||
|
||||
- [ ] **COMMIT (Owner: Brent | Group: M5.tests | Branch: feature/m5-acms-smoke | Planned: Day 28 | Expected: Day 28) - Commit message: "test(e2e): add M5 ACMS + context suites"**
|
||||
- [ ] Git [Brent]: `git checkout master`
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
*** Settings ***
|
||||
Documentation Smoke tests for automation profile guards
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Variables ***
|
||||
${HELPER_SCRIPT} robot/helper_automation_profiles.py
|
||||
|
||||
*** Test Cases ***
|
||||
Guard Allows Tool Within Limits
|
||||
[Documentation] Guard allows tool when under call limit
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-allow cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-allow-ok
|
||||
|
||||
Guard Blocks Tool On Denylist
|
||||
[Documentation] Guard blocks tool that is on the denylist
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-deny cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-deny-ok
|
||||
|
||||
Guard Blocks Exceeding Max Calls
|
||||
[Documentation] Guard blocks when call limit exceeded
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-max-calls cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-max-calls-ok
|
||||
|
||||
Guard Blocks Exceeding Cost Budget
|
||||
[Documentation] Guard blocks when cost budget exceeded
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-cost cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-cost-ok
|
||||
|
||||
Guard Write Approval
|
||||
[Documentation] Guard requires approval for writes
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-write cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-write-ok
|
||||
|
||||
Guard Apply Approval
|
||||
[Documentation] Guard requires approval for apply phase
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} guard-apply cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} guard-apply-ok
|
||||
|
||||
Profile From YAML
|
||||
[Documentation] Load profile with guards from YAML
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} from-yaml cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} from-yaml-ok
|
||||
|
||||
Effective Profile Resolution
|
||||
[Documentation] Effective profile resolution precedence
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} resolution cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} resolution-ok
|
||||
|
||||
Guard Summary
|
||||
[Documentation] Print guard feature summary
|
||||
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} summary cwd=${WORKSPACE}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} summary-ok
|
||||
@@ -0,0 +1,146 @@
|
||||
"""Helper script for Robot Framework automation profile guard tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
import yaml
|
||||
|
||||
from cleveragents.application.services.automation_profile_service import (
|
||||
AutomationProfileService,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
AutomationProfile,
|
||||
)
|
||||
|
||||
|
||||
def _test_guard_allow() -> None:
|
||||
"""Guard allows tool within limits."""
|
||||
g = AutomationGuard(max_tool_calls_per_step=5)
|
||||
p = AutomationProfile(name="test/allow", guards=g)
|
||||
r = p.check_guard("read_file", calls_so_far=3)
|
||||
assert r.allowed is True
|
||||
print("guard-allow-ok")
|
||||
|
||||
|
||||
def _test_guard_deny() -> None:
|
||||
"""Guard blocks tool on denylist."""
|
||||
g = AutomationGuard(tool_denylist=["shell_exec"])
|
||||
p = AutomationProfile(name="test/deny", guards=g)
|
||||
r = p.check_guard("shell_exec")
|
||||
assert r.allowed is False
|
||||
assert "denylist" in (r.reason or "")
|
||||
print("guard-deny-ok")
|
||||
|
||||
|
||||
def _test_guard_max_calls() -> None:
|
||||
"""Guard blocks when call limit exceeded."""
|
||||
g = AutomationGuard(max_tool_calls_per_step=3)
|
||||
p = AutomationProfile(name="test/max-calls", guards=g)
|
||||
r = p.check_guard("tool", calls_so_far=3)
|
||||
assert r.allowed is False
|
||||
assert r.requires_approval is True
|
||||
print("guard-max-calls-ok")
|
||||
|
||||
|
||||
def _test_guard_cost() -> None:
|
||||
"""Guard blocks when cost budget exceeded."""
|
||||
g = AutomationGuard(max_total_cost=10.0)
|
||||
p = AutomationProfile(name="test/cost", guards=g)
|
||||
r = p.check_guard("tool", cost_so_far=10.0)
|
||||
assert r.allowed is False
|
||||
assert "Cost" in (r.reason or "")
|
||||
print("guard-cost-ok")
|
||||
|
||||
|
||||
def _test_guard_write() -> None:
|
||||
"""Guard requires approval for writes."""
|
||||
g = AutomationGuard(require_approval_for_writes=True)
|
||||
p = AutomationProfile(name="test/write", guards=g)
|
||||
r = p.check_guard("write_file", is_write=True)
|
||||
assert r.allowed is False
|
||||
assert r.requires_approval is True
|
||||
print("guard-write-ok")
|
||||
|
||||
|
||||
def _test_guard_apply() -> None:
|
||||
"""Guard requires approval for apply."""
|
||||
g = AutomationGuard(require_approval_for_apply=True)
|
||||
p = AutomationProfile(name="test/apply", guards=g)
|
||||
r = p.check_guard("__apply__")
|
||||
assert r.allowed is False
|
||||
assert r.requires_approval is True
|
||||
print("guard-apply-ok")
|
||||
|
||||
|
||||
def _test_from_yaml() -> None:
|
||||
"""Load profile from YAML."""
|
||||
cfg: dict[str, Any] = {
|
||||
"name": "test/yaml",
|
||||
"guards": {"max_tool_calls_per_step": 5},
|
||||
}
|
||||
tmp_path = tempfile.mktemp(suffix=".yaml")
|
||||
with open(tmp_path, "w") as fh:
|
||||
yaml.dump(cfg, fh)
|
||||
p = AutomationProfile.from_yaml(tmp_path)
|
||||
assert p.guards is not None
|
||||
assert p.guards.max_tool_calls_per_step == 5
|
||||
print("from-yaml-ok")
|
||||
|
||||
|
||||
def _test_resolution() -> None:
|
||||
"""Effective profile resolution precedence."""
|
||||
svc = AutomationProfileService()
|
||||
p1 = svc.get_effective_profile(plan_profile="cautious")
|
||||
assert p1.name == "cautious"
|
||||
p2 = svc.get_effective_profile(action_profile="manual")
|
||||
assert p2.name == "manual"
|
||||
p3 = svc.get_effective_profile(default_profile="auto")
|
||||
assert p3.name == "auto"
|
||||
print("resolution-ok")
|
||||
|
||||
|
||||
def _test_summary() -> None:
|
||||
"""Print guard feature summary."""
|
||||
g = AutomationGuard(
|
||||
max_tool_calls_per_step=10,
|
||||
max_total_cost=50.0,
|
||||
tool_denylist=["dangerous"],
|
||||
require_approval_for_writes=True,
|
||||
require_approval_for_apply=True,
|
||||
)
|
||||
p = AutomationProfile(name="test/summary", guards=g)
|
||||
r1 = p.check_guard("safe_tool")
|
||||
assert r1.allowed is True
|
||||
r2 = p.check_guard("dangerous")
|
||||
assert r2.allowed is False
|
||||
print("summary-ok")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
cmd = sys.argv[1] if len(sys.argv) > 1 else "summary"
|
||||
dispatch: dict[str, Any] = {
|
||||
"guard-allow": _test_guard_allow,
|
||||
"guard-deny": _test_guard_deny,
|
||||
"guard-max-calls": _test_guard_max_calls,
|
||||
"guard-cost": _test_guard_cost,
|
||||
"guard-write": _test_guard_write,
|
||||
"guard-apply": _test_guard_apply,
|
||||
"from-yaml": _test_from_yaml,
|
||||
"resolution": _test_resolution,
|
||||
"summary": _test_summary,
|
||||
}
|
||||
fn = dispatch.get(cmd)
|
||||
if fn:
|
||||
fn()
|
||||
else:
|
||||
print(f"Unknown command: {cmd}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -16,6 +16,7 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from cleveragents.core.exceptions import (
|
||||
@@ -25,6 +26,7 @@ from cleveragents.core.exceptions import (
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationProfile,
|
||||
GuardResult,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -242,3 +244,82 @@ class AutomationProfileService:
|
||||
resource_id=name,
|
||||
)
|
||||
logger.info("Deleted profile '%s'", name)
|
||||
|
||||
# ----- Guard evaluation ----- #
|
||||
|
||||
def evaluate_guard(
|
||||
self,
|
||||
profile_name: str,
|
||||
tool_name: str,
|
||||
context: dict[str, Any] | None = None,
|
||||
) -> GuardResult:
|
||||
"""Evaluate guard constraints for a tool invocation.
|
||||
|
||||
Looks up the profile by *profile_name*, then delegates to
|
||||
:meth:`AutomationProfile.check_guard`.
|
||||
|
||||
Args:
|
||||
profile_name: Name of the automation profile to use.
|
||||
tool_name: Name of the tool being invoked.
|
||||
context: Optional context dict with keys ``is_write``
|
||||
(bool), ``cost_so_far`` (float), ``calls_so_far``
|
||||
(int).
|
||||
|
||||
Returns:
|
||||
A :class:`GuardResult` indicating the decision.
|
||||
|
||||
Raises:
|
||||
NotFoundError: If the profile is not found.
|
||||
"""
|
||||
if not profile_name:
|
||||
raise ValidationError("profile_name must not be empty")
|
||||
if not tool_name:
|
||||
raise ValidationError("tool_name must not be empty")
|
||||
|
||||
profile = self.get_profile(profile_name)
|
||||
ctx = context or {}
|
||||
return profile.check_guard(
|
||||
tool_name=tool_name,
|
||||
is_write=ctx.get("is_write", False),
|
||||
cost_so_far=ctx.get("cost_so_far", 0.0),
|
||||
calls_so_far=ctx.get("calls_so_far", 0),
|
||||
)
|
||||
|
||||
def get_effective_profile(
|
||||
self,
|
||||
plan_profile: str | None = None,
|
||||
action_profile: str | None = None,
|
||||
default_profile: str | None = None,
|
||||
) -> AutomationProfile:
|
||||
"""Resolve the effective profile from plan -> action -> default.
|
||||
|
||||
This is a convenience wrapper around :meth:`resolve_profile`
|
||||
that uses the three-level precedence: plan > action > default.
|
||||
|
||||
Args:
|
||||
plan_profile: Profile name set at plan level.
|
||||
action_profile: Profile name set on the action.
|
||||
default_profile: Fallback default profile name.
|
||||
|
||||
Returns:
|
||||
The resolved ``AutomationProfile``.
|
||||
"""
|
||||
return self.resolve_profile(
|
||||
plan_profile=plan_profile,
|
||||
action_profile=action_profile,
|
||||
project_profile=default_profile,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def from_yaml(path: str | Path) -> AutomationProfile:
|
||||
"""Load an AutomationProfile from a YAML config file.
|
||||
|
||||
Delegates to :meth:`AutomationProfile.from_yaml`.
|
||||
|
||||
Args:
|
||||
path: Path to the YAML configuration file.
|
||||
|
||||
Returns:
|
||||
The loaded ``AutomationProfile``.
|
||||
"""
|
||||
return AutomationProfile.from_yaml(path)
|
||||
|
||||
@@ -1,33 +1,7 @@
|
||||
"""Automation profile management commands for CleverAgents CLI.
|
||||
|
||||
The ``agents automation-profile`` command group manages automation profiles
|
||||
that control how much autonomy the system has during plan execution.
|
||||
|
||||
## Commands
|
||||
|
||||
| Command | Description |
|
||||
|---------------------------------------|------------------------------------|
|
||||
| ``agents automation-profile add`` | Add/register a profile from YAML |
|
||||
| ``agents automation-profile remove`` | Remove a custom profile by name |
|
||||
| ``agents automation-profile list`` | List profiles with optional filters|
|
||||
| ``agents automation-profile show`` | Show full profile details |
|
||||
|
||||
## YAML Configuration File
|
||||
|
||||
.. code-block:: yaml
|
||||
|
||||
name: acme/strict
|
||||
description: Strict review for production
|
||||
schema_version: "1.0"
|
||||
auto_strategize: 1.0
|
||||
auto_execute: 1.0
|
||||
auto_apply: 1.0
|
||||
...
|
||||
|
||||
## Profile Naming
|
||||
|
||||
Built-in profiles use bare names (e.g. ``manual``). Custom profiles
|
||||
use a ``namespace/name`` pattern (e.g. ``acme/strict``).
|
||||
Provides add, remove, list, and show commands for automation profiles
|
||||
that control plan execution autonomy, including guard enforcement hooks.
|
||||
|
||||
Based on v3_spec.md implementation plan Stage A6.cli.
|
||||
"""
|
||||
@@ -58,6 +32,7 @@ from cleveragents.core.exceptions import (
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationGuard,
|
||||
AutomationProfile,
|
||||
)
|
||||
|
||||
@@ -72,23 +47,15 @@ _FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)
|
||||
|
||||
|
||||
class _InMemoryProfileRepository:
|
||||
"""Minimal in-memory repository for custom automation profiles.
|
||||
|
||||
Implements the same interface as ``AutomationProfileRepository`` so
|
||||
``AutomationProfileService`` can be used without a database. This
|
||||
will be replaced by the DB-backed repository once full persistence
|
||||
is wired into the DI container.
|
||||
"""
|
||||
"""In-memory repository for custom automation profiles."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._profiles: dict[str, AutomationProfile] = {}
|
||||
|
||||
def get_by_name(self, name: str) -> AutomationProfile | None:
|
||||
"""Look up a profile by name."""
|
||||
return self._profiles.get(name)
|
||||
|
||||
def list_all(self) -> list[AutomationProfile]:
|
||||
"""Return all persisted profiles ordered by name."""
|
||||
return [self._profiles[k] for k in sorted(self._profiles)]
|
||||
|
||||
def upsert(
|
||||
@@ -96,14 +63,9 @@ class _InMemoryProfileRepository:
|
||||
profile: AutomationProfile,
|
||||
expected_schema_version: str | None = None,
|
||||
) -> None:
|
||||
"""Insert or update a profile."""
|
||||
self._profiles[profile.name] = profile
|
||||
|
||||
def delete(self, name: str) -> None:
|
||||
"""Delete a profile by name.
|
||||
|
||||
Raises ``NotFoundError`` if not present.
|
||||
"""
|
||||
if name not in self._profiles:
|
||||
raise NotFoundError(
|
||||
f"Profile '{name}' not found",
|
||||
@@ -113,27 +75,29 @@ class _InMemoryProfileRepository:
|
||||
del self._profiles[name]
|
||||
|
||||
|
||||
# Module-level in-memory repo so profiles persist within a CLI session
|
||||
_repo = _InMemoryProfileRepository()
|
||||
|
||||
|
||||
def _get_service() -> AutomationProfileService:
|
||||
"""Get the AutomationProfileService with in-memory repository.
|
||||
|
||||
Uses a module-level in-memory repository. When the full DI
|
||||
container is wired this helper will be updated to use the
|
||||
container instead.
|
||||
|
||||
The in-memory repo implements the same interface as
|
||||
``AutomationProfileRepository`` and is duck-type compatible.
|
||||
"""
|
||||
# The in-memory repo is duck-type compatible with AutomationProfileRepository.
|
||||
# Using Any cast for the type checker since the formal type is only
|
||||
# available behind TYPE_CHECKING (to avoid importing SQLAlchemy).
|
||||
"""Get the AutomationProfileService with in-memory repository."""
|
||||
repo: Any = _repo
|
||||
return AutomationProfileService(repo=repo)
|
||||
|
||||
|
||||
def _guards_dict(guards: AutomationGuard | None) -> dict[str, object] | None:
|
||||
"""Return guard data as a dict suitable for CLI output."""
|
||||
if guards is None:
|
||||
return None
|
||||
return {
|
||||
"max_tool_calls_per_step": guards.max_tool_calls_per_step,
|
||||
"max_total_cost": guards.max_total_cost,
|
||||
"tool_allowlist": guards.tool_allowlist,
|
||||
"tool_denylist": guards.tool_denylist,
|
||||
"require_approval_for_writes": guards.require_approval_for_writes,
|
||||
"require_approval_for_apply": guards.require_approval_for_apply,
|
||||
}
|
||||
|
||||
|
||||
def _profile_spec_dict(profile: AutomationProfile) -> dict[str, object]:
|
||||
"""Return profile data as a dict suitable for CLI output.
|
||||
|
||||
@@ -160,6 +124,7 @@ def _profile_spec_dict(profile: AutomationProfile) -> dict[str, object]:
|
||||
"require_sandbox": profile.require_sandbox,
|
||||
"require_checkpoints": profile.require_checkpoints,
|
||||
"allow_unsafe_tools": profile.allow_unsafe_tools,
|
||||
"guards": _guards_dict(profile.guards),
|
||||
}
|
||||
return result
|
||||
|
||||
@@ -212,6 +177,20 @@ def _print_profile(
|
||||
f" allow_unsafe_tools: {profile.allow_unsafe_tools}"
|
||||
)
|
||||
|
||||
if profile.guards is not None:
|
||||
g = profile.guards
|
||||
details += (
|
||||
f"\n\n[bold]Guards:[/bold]\n"
|
||||
f" max_tool_calls_per_step: {g.max_tool_calls_per_step}\n"
|
||||
f" max_total_cost: {g.max_total_cost}\n"
|
||||
f" tool_allowlist: {g.tool_allowlist}\n"
|
||||
f" tool_denylist: {g.tool_denylist}\n"
|
||||
f" require_approval_for_writes: "
|
||||
f"{g.require_approval_for_writes}\n"
|
||||
f" require_approval_for_apply: "
|
||||
f"{g.require_approval_for_apply}"
|
||||
)
|
||||
|
||||
console.print(Panel(details, title=title, expand=False))
|
||||
|
||||
|
||||
@@ -220,37 +199,19 @@ def add_profile(
|
||||
config: Annotated[
|
||||
Path,
|
||||
typer.Option(
|
||||
"--config",
|
||||
"-c",
|
||||
help="Path to the automation profile YAML configuration file",
|
||||
exists=False,
|
||||
"--config", "-c", help="Path to profile YAML config", exists=False
|
||||
),
|
||||
],
|
||||
update: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
"--update",
|
||||
help="Update an existing profile instead of creating a new one",
|
||||
),
|
||||
typer.Option("--update", help="Update existing profile"),
|
||||
] = False,
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
"--format",
|
||||
"-f",
|
||||
help=_FORMAT_HELP,
|
||||
),
|
||||
typer.Option("--format", "-f", help=_FORMAT_HELP),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Add or register an automation profile from a YAML configuration file.
|
||||
|
||||
Profiles are created via ``--config <file>``. Use ``--update`` to
|
||||
overwrite an existing custom profile.
|
||||
|
||||
Examples:
|
||||
agents automation-profile add --config ./profiles/strict.yaml
|
||||
agents automation-profile add --config ./profiles/strict.yaml --update
|
||||
"""
|
||||
"""Add or register an automation profile from a YAML config file."""
|
||||
try:
|
||||
if not config.exists():
|
||||
console.print(f"[red]Config file error:[/red] File not found: {config}")
|
||||
@@ -325,33 +286,18 @@ def add_profile(
|
||||
def remove_profile(
|
||||
name: Annotated[
|
||||
str,
|
||||
typer.Argument(help="Name of the automation profile to remove"),
|
||||
typer.Argument(help="Profile name to remove"),
|
||||
],
|
||||
yes: Annotated[
|
||||
bool,
|
||||
typer.Option(
|
||||
"--yes",
|
||||
"-y",
|
||||
help="Skip confirmation prompt",
|
||||
),
|
||||
typer.Option("--yes", "-y", help="Skip confirmation"),
|
||||
] = False,
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option(
|
||||
"--format",
|
||||
"-f",
|
||||
help=_FORMAT_HELP,
|
||||
),
|
||||
typer.Option("--format", "-f", help=_FORMAT_HELP),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Remove a custom automation profile by name.
|
||||
|
||||
Built-in profiles cannot be removed.
|
||||
|
||||
Examples:
|
||||
agents automation-profile remove acme/strict
|
||||
agents automation-profile remove acme/strict --yes
|
||||
"""
|
||||
"""Remove a custom automation profile by name."""
|
||||
try:
|
||||
service = _get_service()
|
||||
|
||||
|
||||
@@ -3,7 +3,9 @@ from cleveragents.domain.models.core.action import ActionState
|
||||
from cleveragents.domain.models.core.actor import Actor
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationGuard,
|
||||
AutomationProfile,
|
||||
GuardResult,
|
||||
get_builtin_profile,
|
||||
)
|
||||
from cleveragents.domain.models.core.change import (
|
||||
@@ -149,6 +151,7 @@ __all__ = [
|
||||
"BUILTIN_PROFILES",
|
||||
"ActionState",
|
||||
"Actor",
|
||||
"AutomationGuard",
|
||||
"AutomationProfile",
|
||||
"BindingMode",
|
||||
"BindingResult",
|
||||
@@ -170,6 +173,7 @@ __all__ = [
|
||||
"CreditsTransaction",
|
||||
"CreditsTransactionType",
|
||||
"DebugAttempt",
|
||||
"GuardResult",
|
||||
"InMemoryChangeSetStore",
|
||||
"InMemoryInvocationTracker",
|
||||
"InvariantSource",
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Automation Guard models for CleverAgents v3.
|
||||
|
||||
An **AutomationGuard** provides runtime enforcement hooks that gate tool
|
||||
invocations beyond the phase-transition thresholds defined by an
|
||||
:class:`AutomationProfile`. Guards support:
|
||||
|
||||
- **Max tool calls** — Limit the number of tool invocations per step.
|
||||
- **Budget cap** — Halt automatic execution when cumulative cost
|
||||
exceeds a threshold.
|
||||
- **Allowlist / Denylist** — Control which tools may proceed
|
||||
automatically.
|
||||
- **Write approval** — Require human approval for write operations.
|
||||
- **Apply approval** — Require human approval before the apply phase.
|
||||
|
||||
A **GuardResult** captures the outcome of a guard evaluation:
|
||||
``allowed``, ``reason``, and ``requires_approval``.
|
||||
|
||||
Based on ``docs/specification.md`` Section "Automation Profiles"
|
||||
(enforcement hooks).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GuardResult
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class GuardResult(BaseModel):
|
||||
"""Result of evaluating an automation guard check.
|
||||
|
||||
Returned by :meth:`AutomationProfile.check_guard` to indicate
|
||||
whether a tool invocation is allowed, blocked, or requires human
|
||||
approval.
|
||||
"""
|
||||
|
||||
allowed: bool = Field(
|
||||
...,
|
||||
description="Whether the tool invocation is allowed to proceed",
|
||||
)
|
||||
reason: str | None = Field(
|
||||
default=None,
|
||||
description="Human-readable reason when not allowed",
|
||||
)
|
||||
requires_approval: bool = Field(
|
||||
default=False,
|
||||
description=("Whether human approval is required before proceeding"),
|
||||
)
|
||||
|
||||
model_config = ConfigDict(
|
||||
str_strip_whitespace=True,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AutomationGuard
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class AutomationGuard(BaseModel):
|
||||
"""Enforcement hooks for an automation profile.
|
||||
|
||||
Guards provide additional runtime constraints beyond the phase-
|
||||
transition thresholds. They gate tool invocations based on call
|
||||
counts, budgets, allowlists/denylists, and write/apply semantics.
|
||||
"""
|
||||
|
||||
max_tool_calls_per_step: int | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Maximum tool invocations per step before requiring "
|
||||
"approval. None means unlimited."
|
||||
),
|
||||
)
|
||||
max_total_cost: float | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Budget cap (cumulative cost) before requiring approval. "
|
||||
"None means unlimited."
|
||||
),
|
||||
)
|
||||
tool_allowlist: list[str] | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Only these tools may be called automatically. "
|
||||
"None means all tools are allowed."
|
||||
),
|
||||
)
|
||||
tool_denylist: list[str] | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"These tools always require human approval. "
|
||||
"None means no tools are denied."
|
||||
),
|
||||
)
|
||||
require_approval_for_writes: bool = Field(
|
||||
default=False,
|
||||
description="Require human approval for write operations",
|
||||
)
|
||||
require_approval_for_apply: bool = Field(
|
||||
default=False,
|
||||
description="Require human approval before the apply phase",
|
||||
)
|
||||
|
||||
@field_validator("max_tool_calls_per_step")
|
||||
@classmethod
|
||||
def validate_max_tool_calls(
|
||||
cls: type[AutomationGuard],
|
||||
v: int | None,
|
||||
) -> int | None:
|
||||
"""Validate max_tool_calls_per_step is positive if set."""
|
||||
if v is not None and v < 0:
|
||||
raise ValueError(f"max_tool_calls_per_step must be >= 0, got {v}")
|
||||
return v
|
||||
|
||||
@field_validator("max_total_cost")
|
||||
@classmethod
|
||||
def validate_max_total_cost(
|
||||
cls: type[AutomationGuard],
|
||||
v: float | None,
|
||||
) -> float | None:
|
||||
"""Validate max_total_cost is non-negative if set."""
|
||||
if v is not None and v < 0.0:
|
||||
raise ValueError(f"max_total_cost must be >= 0.0, got {v}")
|
||||
return v
|
||||
|
||||
model_config = ConfigDict(
|
||||
str_strip_whitespace=True,
|
||||
validate_assignment=True,
|
||||
)
|
||||
@@ -1,45 +1,27 @@
|
||||
"""Automation Profile domain model for CleverAgents v3.
|
||||
|
||||
An **AutomationProfile** defines a set of threshold values that control
|
||||
how much autonomy the system has at each phase of plan execution. Each
|
||||
threshold is a float in [0.0, 1.0] where:
|
||||
Defines threshold values controlling plan-execution autonomy.
|
||||
Each threshold is a float in [0.0, 1.0]: 0.0 = automatic,
|
||||
1.0 = requires human approval. Eight built-in profiles ship
|
||||
with every installation. Custom profiles use ``namespace/name``.
|
||||
|
||||
- **0.0** means the action proceeds automatically (no human gate).
|
||||
- **1.0** means human approval is always required before proceeding.
|
||||
- Values in between represent a confidence threshold: the system may
|
||||
proceed autonomously only when its confidence exceeds the threshold.
|
||||
|
||||
## Built-in Profiles
|
||||
|
||||
Eight built-in profiles ship with every installation:
|
||||
|
||||
| Profile | Intent |
|
||||
|--------------|--------------------------------------------|
|
||||
| ``manual`` | Human approves every action |
|
||||
| ``review`` | Human reviews before apply |
|
||||
| ``supervised``| Human reviews strategy + execution |
|
||||
| ``cautious`` | Probabilistic gates on most actions |
|
||||
| ``trusted`` | Auto for most, human for apply + revert |
|
||||
| ``auto`` | Fully automatic except reversion |
|
||||
| ``ci`` | Designed for CI pipelines |
|
||||
| ``full-auto``| No gates, no sandbox, no checkpoints |
|
||||
|
||||
## Profile Naming
|
||||
|
||||
Built-in profiles use bare names (e.g. ``manual``). Custom profiles
|
||||
use a ``namespace/name`` pattern (e.g. ``acme/strict``).
|
||||
|
||||
Based on ``docs/specification.md`` Section "Automation Profiles"
|
||||
(lines 13424-13612).
|
||||
Based on ``docs/specification.md`` Section "Automation Profiles".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
GuardResult,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regex patterns
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -176,6 +158,13 @@ class AutomationProfile(BaseModel):
|
||||
description="Whether unsafe tools may be used",
|
||||
)
|
||||
|
||||
# -- Guards ------------------------------------------------------------
|
||||
|
||||
guards: AutomationGuard | None = Field(
|
||||
default=None,
|
||||
description="Optional enforcement hooks for runtime constraints",
|
||||
)
|
||||
|
||||
# -- Name validation ---------------------------------------------------
|
||||
|
||||
@field_validator("name")
|
||||
@@ -227,6 +216,69 @@ class AutomationProfile(BaseModel):
|
||||
raise ValueError(f"Threshold must be between 0.0 and 1.0, got {v}")
|
||||
return v
|
||||
|
||||
# -- Guard checking ----------------------------------------------------
|
||||
|
||||
def check_guard(
|
||||
self,
|
||||
tool_name: str,
|
||||
is_write: bool = False,
|
||||
cost_so_far: float = 0.0,
|
||||
calls_so_far: int = 0,
|
||||
) -> GuardResult:
|
||||
"""Evaluate guard constraints for a tool invocation."""
|
||||
if tool_name is None:
|
||||
raise ValueError("tool_name cannot be None")
|
||||
if not isinstance(tool_name, str):
|
||||
raise TypeError("tool_name must be a string")
|
||||
if cost_so_far < 0.0:
|
||||
raise ValueError(f"cost_so_far must be >= 0.0, got {cost_so_far}")
|
||||
if calls_so_far < 0:
|
||||
raise ValueError(f"calls_so_far must be >= 0, got {calls_so_far}")
|
||||
if self.guards is None:
|
||||
return GuardResult(allowed=True)
|
||||
guards = self.guards
|
||||
if guards.tool_denylist and tool_name in guards.tool_denylist:
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Tool '{tool_name}' is on the denylist",
|
||||
)
|
||||
if guards.tool_allowlist is not None and tool_name not in guards.tool_allowlist:
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Tool '{tool_name}' is not on the allowlist",
|
||||
)
|
||||
if (
|
||||
guards.max_tool_calls_per_step is not None
|
||||
and calls_so_far >= guards.max_tool_calls_per_step
|
||||
):
|
||||
limit = guards.max_tool_calls_per_step
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Tool call limit reached ({limit})",
|
||||
)
|
||||
if guards.max_total_cost is not None and cost_so_far >= guards.max_total_cost:
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Cost budget exceeded ({guards.max_total_cost})",
|
||||
)
|
||||
if guards.require_approval_for_writes and is_write:
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason="Write operations require approval",
|
||||
)
|
||||
if guards.require_approval_for_apply and tool_name == "__apply__":
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason="Apply phase requires approval",
|
||||
)
|
||||
return GuardResult(allowed=True)
|
||||
|
||||
# -- Factories ---------------------------------------------------------
|
||||
|
||||
@classmethod
|
||||
@@ -236,6 +288,29 @@ class AutomationProfile(BaseModel):
|
||||
raise ValueError("AutomationProfile config must include 'name'")
|
||||
return cls.model_validate(config)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, path: str | Path) -> AutomationProfile:
|
||||
"""Create an AutomationProfile from a YAML file.
|
||||
|
||||
Args:
|
||||
path: Path to the YAML configuration file.
|
||||
|
||||
Returns:
|
||||
The loaded ``AutomationProfile``.
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If the file does not exist.
|
||||
ValueError: If the YAML content is invalid.
|
||||
"""
|
||||
file_path = Path(path)
|
||||
if not file_path.exists():
|
||||
raise FileNotFoundError(f"Profile YAML file not found: {file_path}")
|
||||
with open(file_path) as fh:
|
||||
config = yaml.safe_load(fh)
|
||||
if not isinstance(config, dict):
|
||||
raise ValueError("Profile YAML must contain a mapping (dict)")
|
||||
return cls.from_config(config)
|
||||
|
||||
model_config = ConfigDict(
|
||||
str_strip_whitespace=True,
|
||||
validate_assignment=True,
|
||||
|
||||
@@ -115,6 +115,22 @@ _LEVEL_TO_PROFILE # noqa: B018, F821
|
||||
_resolve_profile_for_plan # noqa: B018, F821
|
||||
default_automation_profile # noqa: B018, F821
|
||||
|
||||
# Automation guards — public API (M5.3)
|
||||
AutomationGuard # noqa: B018, F821
|
||||
GuardResult # noqa: B018, F821
|
||||
check_guard # noqa: B018, F821
|
||||
evaluate_guard # noqa: B018, F821
|
||||
get_effective_profile # noqa: B018, F821
|
||||
from_yaml # noqa: B018, F821
|
||||
max_tool_calls_per_step # noqa: B018, F821
|
||||
max_total_cost # noqa: B018, F821
|
||||
tool_allowlist # noqa: B018, F821
|
||||
tool_denylist # noqa: B018, F821
|
||||
require_approval_for_writes # noqa: B018, F821
|
||||
require_approval_for_apply # noqa: B018, F821
|
||||
requires_approval # noqa: B018, F821
|
||||
guards # noqa: B018, F821
|
||||
|
||||
# Config security scanner — public API (SEC1)
|
||||
Severity # noqa: B018, F821
|
||||
Violation # noqa: B018, F821
|
||||
|
||||
Reference in New Issue
Block a user