feat(autonomy): guard enforcement works (denylist, budget caps, tool call limits) #1204
@@ -802,3 +802,26 @@ Feature: Consolidated Automation Profile
|
||||
When I try to check guard with negative calls
|
||||
Then a guard argument error should be raised
|
||||
|
||||
Scenario: Guard denylist message includes remediation guidance
|
||||
Given a profile with denylist "shell_exec"
|
||||
When I check guard for tool "shell_exec"
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "Remediation"
|
||||
|
||||
Scenario: Guard budget message includes remediation guidance
|
||||
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 "Remediation"
|
||||
|
||||
Scenario: Guard tool-call limit can be evaluated per subplan scope
|
||||
Given a profile with max_tool_calls_per_step 2
|
||||
When I check guard for tool "read_file" with 2 calls so far in "subplan" scope
|
||||
Then the guard result should not be allowed
|
||||
And the guard reason should mention "Subplan"
|
||||
|
||||
Scenario: Guard write approval message includes remediation guidance
|
||||
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 "Remediation"
|
||||
|
||||
@@ -15,6 +15,7 @@ from cleveragents.core.exceptions import ValidationError as CAValidationError
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
AutomationGuard,
|
||||
GuardResult,
|
||||
GuardScope,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
AutomationProfile,
|
||||
@@ -92,6 +93,20 @@ def step_check_guard_calls(
|
||||
)
|
||||
|
||||
|
||||
@when('I check guard for tool "{tool}" with {calls:d} calls so far in "{scope}" scope')
|
||||
def step_check_guard_calls_with_scope(
|
||||
context: Context,
|
||||
tool: str,
|
||||
calls: int,
|
||||
scope: str,
|
||||
) -> None:
|
||||
context.guard_result = context.guard_profile.check_guard(
|
||||
tool_name=tool,
|
||||
calls_so_far=calls,
|
||||
scope=GuardScope(scope),
|
||||
)
|
||||
|
||||
|
||||
@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(
|
||||
|
||||
@@ -23,6 +23,7 @@ from cleveragents.core.exceptions import (
|
||||
NotFoundError,
|
||||
ValidationError,
|
||||
)
|
||||
from cleveragents.domain.models.core.automation_guard import GuardScope
|
||||
from cleveragents.domain.models.core.automation_profile import (
|
||||
BUILTIN_PROFILES,
|
||||
AutomationProfile,
|
||||
@@ -263,7 +264,7 @@ class AutomationProfileService:
|
||||
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).
|
||||
(int), and ``scope`` (:class:`GuardScope`).
|
||||
|
||||
Returns:
|
||||
A :class:`GuardResult` indicating the decision.
|
||||
@@ -283,6 +284,7 @@ class AutomationProfileService:
|
||||
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),
|
||||
scope=ctx.get("scope", GuardScope.PLAN),
|
||||
)
|
||||
|
||||
def get_effective_profile(
|
||||
|
||||
@@ -17,6 +17,7 @@ from cleveragents.domain.models.core.automation_profile import (
|
||||
AutomationGuard,
|
||||
AutomationProfile,
|
||||
GuardResult,
|
||||
GuardScope,
|
||||
get_builtin_profile,
|
||||
)
|
||||
from cleveragents.domain.models.core.autonomy_guardrails import (
|
||||
@@ -402,6 +403,7 @@ __all__ = [
|
||||
"FileRecord",
|
||||
"FragmentProvenance",
|
||||
"GuardResult",
|
||||
"GuardScope",
|
||||
"GuardrailAuditEntry",
|
||||
"GuardrailAuditTrail",
|
||||
"GuardrailEventType",
|
||||
|
||||
@@ -21,8 +21,51 @@ Based on ``docs/specification.md`` Section "Automation Profiles"
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import StrEnum
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GuardScope
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class GuardScope(StrEnum):
|
||||
"""Scope at which a guard check is evaluated.
|
||||
|
||||
Used by :meth:`AutomationProfile.check_guard` to distinguish
|
||||
plan-level from subplan-level enforcement.
|
||||
"""
|
||||
|
||||
PLAN = "plan"
|
||||
SUBPLAN = "subplan"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Remediation guidance constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
REMEDIATION_DENYLIST: str = (
|
||||
"Remediation: remove it from the denylist or request human "
|
||||
"approval for this operation."
|
||||
)
|
||||
REMEDIATION_ALLOWLIST: str = (
|
||||
"Remediation: add it to the allowlist or request human approval."
|
||||
)
|
||||
REMEDIATION_TOOL_CALL_LIMIT: str = (
|
||||
"Remediation: increase max_tool_calls_per_step, reduce tool "
|
||||
"usage, or request human approval."
|
||||
)
|
||||
REMEDIATION_BUDGET: str = (
|
||||
"Remediation: raise max_total_cost, lower cost, or request human approval."
|
||||
)
|
||||
REMEDIATION_WRITE_APPROVAL: str = (
|
||||
"Remediation: request human approval before continuing."
|
||||
)
|
||||
REMEDIATION_APPLY_APPROVAL: str = (
|
||||
"Remediation: request human approval to proceed with apply."
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GuardResult
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -56,8 +56,15 @@ import yaml
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
|
||||
from cleveragents.domain.models.core.automation_guard import (
|
||||
REMEDIATION_ALLOWLIST,
|
||||
REMEDIATION_APPLY_APPROVAL,
|
||||
REMEDIATION_BUDGET,
|
||||
REMEDIATION_DENYLIST,
|
||||
REMEDIATION_TOOL_CALL_LIMIT,
|
||||
REMEDIATION_WRITE_APPROVAL,
|
||||
AutomationGuard,
|
||||
GuardResult,
|
||||
GuardScope,
|
||||
)
|
||||
from cleveragents.domain.models.core.safety_profile import (
|
||||
DEFAULT_SAFETY_PROFILE,
|
||||
@@ -273,6 +280,7 @@ class AutomationProfile(BaseModel):
|
||||
is_write: bool = False,
|
||||
cost_so_far: float = 0.0,
|
||||
calls_so_far: int = 0,
|
||||
scope: GuardScope = GuardScope.PLAN,
|
||||
) -> GuardResult:
|
||||
"""Evaluate guard constraints for a tool invocation."""
|
||||
if tool_name is None:
|
||||
@@ -290,13 +298,19 @@ class AutomationProfile(BaseModel):
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Tool '{tool_name}' is on the denylist",
|
||||
reason=(
|
||||
f"Tool '{tool_name}' is on the denylist for this "
|
||||
f"{scope.value}. {REMEDIATION_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",
|
||||
reason=(
|
||||
f"Tool '{tool_name}' is not on the allowlist for this "
|
||||
f"{scope.value}. {REMEDIATION_ALLOWLIST}"
|
||||
),
|
||||
)
|
||||
if (
|
||||
guards.max_tool_calls_per_step is not None
|
||||
@@ -306,25 +320,35 @@ class AutomationProfile(BaseModel):
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason=f"Tool call limit reached ({limit})",
|
||||
reason=(
|
||||
f"{scope.value.capitalize()} tool call limit reached "
|
||||
f"({calls_so_far}/{limit}). {REMEDIATION_TOOL_CALL_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})",
|
||||
reason=(
|
||||
f"Cost budget exceeded for this {scope.value} "
|
||||
f"({cost_so_far:.2f} >= {guards.max_total_cost:.2f}). "
|
||||
f"{REMEDIATION_BUDGET}"
|
||||
),
|
||||
)
|
||||
if guards.require_approval_for_writes and is_write:
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason="Write operations require approval",
|
||||
reason=(
|
||||
f"Write operations require approval in this "
|
||||
f"{scope.value}. {REMEDIATION_WRITE_APPROVAL}"
|
||||
),
|
||||
)
|
||||
if guards.require_approval_for_apply and tool_name == "__apply__":
|
||||
return GuardResult(
|
||||
allowed=False,
|
||||
requires_approval=True,
|
||||
reason="Apply phase requires approval",
|
||||
reason=(f"Apply phase requires approval. {REMEDIATION_APPLY_APPROVAL}"),
|
||||
)
|
||||
return GuardResult(allowed=True)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user