feat(security): add safety profile enforcement #518
@@ -0,0 +1,190 @@
|
||||
"""ASV benchmarks for Safety Profile domain model operations.
|
||||
|
||||
Measures the performance of:
|
||||
- SafetyProfile construction (Pydantic validation)
|
||||
- SafetyProfile serialization (model_dump / model_validate)
|
||||
- resolve_safety_profile precedence resolution
|
||||
- SafetyProfileProvenance enum operations
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cleveragents.domain.models.core.safety_profile import (
|
||||
DEFAULT_SAFETY_PROFILE,
|
||||
SafetyProfile,
|
||||
SafetyProfileProvenance,
|
||||
resolve_safety_profile,
|
||||
)
|
||||
except ModuleNotFoundError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
from cleveragents.domain.models.core.safety_profile import (
|
||||
DEFAULT_SAFETY_PROFILE,
|
||||
SafetyProfile,
|
||||
SafetyProfileProvenance,
|
||||
resolve_safety_profile,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_profile(**kwargs: object) -> SafetyProfile:
|
||||
"""Create a SafetyProfile with sensible defaults for benchmarks."""
|
||||
return SafetyProfile(**kwargs) # type: ignore[arg-type]
|
||||
|
||||
|
||||
_FULL_CONFIG = {
|
||||
"allowed_skill_categories": ["code", "test", "deploy"],
|
||||
"require_sandbox": False,
|
||||
"require_checkpoints": False,
|
||||
"require_human_approval": True,
|
||||
"allow_unsafe_tools": True,
|
||||
"max_cost_per_plan": 100.0,
|
||||
"max_retries_per_step": 5,
|
||||
"max_total_cost": 500.0,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Construction suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ConstructionSuite:
|
||||
"""Benchmark SafetyProfile construction."""
|
||||
|
||||
def time_default_construction(self) -> None:
|
||||
"""Benchmark creating a default SafetyProfile."""
|
||||
SafetyProfile()
|
||||
|
||||
def time_full_construction(self) -> None:
|
||||
"""Benchmark creating a fully-populated SafetyProfile."""
|
||||
SafetyProfile(
|
||||
allowed_skill_categories=["code", "test", "deploy"],
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
require_human_approval=True,
|
||||
allow_unsafe_tools=True,
|
||||
max_cost_per_plan=100.0,
|
||||
max_retries_per_step=5,
|
||||
max_total_cost=500.0,
|
||||
)
|
||||
|
||||
def time_from_config(self) -> None:
|
||||
"""Benchmark SafetyProfile.from_config factory."""
|
||||
SafetyProfile.from_config(_FULL_CONFIG)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialization suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class SerializationSuite:
|
||||
"""Benchmark SafetyProfile serialization round-trips."""
|
||||
|
||||
def setup(self) -> None:
|
||||
"""Create profiles for serialization benchmarks."""
|
||||
self.profile = _make_profile(
|
||||
allowed_skill_categories=["code", "test"],
|
||||
max_cost_per_plan=50.0,
|
||||
max_total_cost=200.0,
|
||||
)
|
||||
self.dump = self.profile.model_dump()
|
||||
|
||||
def time_model_dump(self) -> None:
|
||||
"""Benchmark model_dump serialization."""
|
||||
self.profile.model_dump()
|
||||
|
||||
def time_model_dump_json(self) -> None:
|
||||
"""Benchmark model_dump_json serialization."""
|
||||
self.profile.model_dump_json()
|
||||
|
||||
def time_model_validate(self) -> None:
|
||||
"""Benchmark model_validate deserialization."""
|
||||
SafetyProfile.model_validate(self.dump)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resolution suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ResolutionSuite:
|
||||
"""Benchmark resolve_safety_profile precedence resolution."""
|
||||
|
||||
def setup(self) -> None:
|
||||
"""Create profiles for resolution benchmarks."""
|
||||
self.plan = SafetyProfile(allow_unsafe_tools=True, require_sandbox=False)
|
||||
self.action = SafetyProfile(allow_unsafe_tools=False)
|
||||
self.project = SafetyProfile(max_cost_per_plan=100.0)
|
||||
self.global_ = SafetyProfile()
|
||||
|
||||
def time_resolve_all_levels(self) -> None:
|
||||
"""Benchmark resolution with all four levels populated."""
|
||||
resolve_safety_profile(
|
||||
plan_profile=self.plan,
|
||||
action_profile=self.action,
|
||||
project_profile=self.project,
|
||||
global_profile=self.global_,
|
||||
)
|
||||
|
||||
def time_resolve_global_only(self) -> None:
|
||||
"""Benchmark resolution with only global level."""
|
||||
resolve_safety_profile(global_profile=self.global_)
|
||||
|
||||
def time_resolve_none(self) -> None:
|
||||
"""Benchmark resolution with all None (falls back to default)."""
|
||||
resolve_safety_profile()
|
||||
|
||||
def time_resolve_action_level(self) -> None:
|
||||
"""Benchmark resolution with action and lower levels."""
|
||||
resolve_safety_profile(
|
||||
action_profile=self.action,
|
||||
project_profile=self.project,
|
||||
global_profile=self.global_,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Enum suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class DefaultProfileSuite:
|
||||
"""Benchmark DEFAULT_SAFETY_PROFILE operations."""
|
||||
|
||||
def time_default_access(self) -> None:
|
||||
"""Benchmark accessing the DEFAULT_SAFETY_PROFILE constant."""
|
||||
_ = DEFAULT_SAFETY_PROFILE
|
||||
|
||||
def time_default_model_dump(self) -> None:
|
||||
"""Benchmark serializing the default profile."""
|
||||
DEFAULT_SAFETY_PROFILE.model_dump()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Enum suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ProvenanceEnumSuite:
|
||||
"""Benchmark SafetyProfileProvenance enum operations."""
|
||||
|
||||
def time_enum_access(self) -> None:
|
||||
"""Benchmark direct enum member access."""
|
||||
_ = SafetyProfileProvenance.PLAN
|
||||
|
||||
def time_enum_from_value(self) -> None:
|
||||
"""Benchmark enum construction from string value."""
|
||||
SafetyProfileProvenance("plan")
|
||||
|
||||
def time_enum_iteration(self) -> None:
|
||||
"""Benchmark enum iteration."""
|
||||
list(SafetyProfileProvenance)
|
||||
@@ -0,0 +1,163 @@
|
||||
# Safety Profiles
|
||||
|
||||
Safety profiles control hard safety constraints for plan execution. Unlike
|
||||
autonomy thresholds (which are confidence-dependent), safety constraints are
|
||||
binary invariants enforced regardless of the system's confidence level.
|
||||
|
||||
## Overview
|
||||
|
||||
A `SafetyProfile` is a composed sub-model of `AutomationProfile` that groups
|
||||
all hard safety constraints. It can also be attached directly to an `Action`
|
||||
when only safety constraints are needed without full autonomy thresholds.
|
||||
|
||||
## Fields
|
||||
|
||||
| Field | Type | Default | Description |
|
||||
|-------|------|---------|-------------|
|
||||
| `require_sandbox` | `bool` | `true` | Require sandbox isolation for execution |
|
||||
| `require_checkpoints` | `bool` | `true` | Require checkpointing before writes |
|
||||
| `allow_unsafe_tools` | `bool` | `false` | Allow tools marked `unsafe` |
|
||||
| `require_human_approval` | `bool` | `false` | Require human approval per action step |
|
||||
| `allowed_skill_categories` | `list[str]` | `[]` | Skill categories allowed (empty = all) |
|
||||
| `max_cost_per_plan` | `float \| None` | `None` | Max cost in USD per plan |
|
||||
| `max_retries_per_step` | `int` | `3` | Max retries per action step (0-100) |
|
||||
| `max_total_cost` | `float \| None` | `None` | Max total cost across all plans |
|
||||
|
||||
## Resolution Precedence
|
||||
|
||||
Safety profiles are resolved at `plan use` time using this precedence
|
||||
(highest to lowest):
|
||||
|
||||
1. **Plan-level** -- set via `--automation-profile` on `agents plan use`
|
||||
2. **Action-level** -- set on the action via `--automation-profile`
|
||||
3. **Project-level** -- set via `agents config set core.automation-profile`
|
||||
4. **Global-level** -- set via `agents config set core.automation-profile`
|
||||
|
||||
Once resolved, the profile is **locked to the plan** -- subsequent changes
|
||||
to project or global profiles do not affect running plans.
|
||||
|
||||
When all levels are `None`, the `DEFAULT_SAFETY_PROFILE` is returned with
|
||||
`GLOBAL` provenance.
|
||||
|
||||
```python
|
||||
from cleveragents.domain.models.core.safety_profile import (
|
||||
SafetyProfile,
|
||||
resolve_safety_profile,
|
||||
)
|
||||
|
||||
plan_safety = SafetyProfile(allow_unsafe_tools=True, require_sandbox=False)
|
||||
resolved, provenance = resolve_safety_profile(plan_profile=plan_safety)
|
||||
# provenance == SafetyProfileProvenance.PLAN
|
||||
```
|
||||
|
||||
## Enforcement
|
||||
|
||||
Safety profile constraints are enforced at tool activation and execution
|
||||
time in `ToolRuntime._enforce_capabilities()`. The checks are evaluated
|
||||
in order; the first failing check raises the corresponding error.
|
||||
|
||||
### 1. Read-Only Plans
|
||||
|
||||
Read-only plan enforcement (`plan_read_only`) is independent of the safety
|
||||
profile and always applies: tools with `writes=True` are blocked with
|
||||
`ToolAccessDeniedError`.
|
||||
|
||||
### 2. Checkpoint Requirements
|
||||
|
||||
When `require_checkpoints=True` (from either the context flag or the safety
|
||||
profile), tools without `capability.checkpointable=True` are blocked with
|
||||
a `ToolCheckpointRequiredError`.
|
||||
|
||||
### 3. Unsafe Tool Gating
|
||||
|
||||
Tools with `capability.unsafe=True` are blocked unless the safety profile
|
||||
has `allow_unsafe_tools=True`. When blocked, a `ToolSafetyViolationError`
|
||||
is raised.
|
||||
|
||||
### 4. Skill Category Allow-List
|
||||
|
||||
When `allowed_skill_categories` is non-empty, only tools belonging to a
|
||||
listed category may execute. The tool's skill category is carried in
|
||||
`ToolExecutionContext.metadata["tool_skill_category"]`. An empty list means
|
||||
all categories are allowed. If the metadata key is missing when the
|
||||
allow-list is non-empty, a `ToolSafetyViolationError` is raised with a
|
||||
clear message indicating the missing metadata.
|
||||
|
||||
### 5. Sandbox Requirement
|
||||
|
||||
When `require_sandbox=True`, tools with `capability.writes=True` require a
|
||||
`sandbox_id` to be set on the `ToolExecutionContext`. If no sandbox is
|
||||
available, a `ToolSandboxRequiredError` is raised. Read-only tools are not
|
||||
affected by this check since they do not modify resources.
|
||||
|
||||
### 6. Human Approval
|
||||
|
||||
When `require_human_approval=True`, all tool executions require that
|
||||
`ToolExecutionContext.metadata["human_approved"]` is set to `True`. If
|
||||
approval has not been recorded, a `ToolHumanApprovalRequiredError` is
|
||||
raised. The higher-level orchestrator is responsible for obtaining approval
|
||||
and setting this metadata before tool execution.
|
||||
|
||||
### 7. Cost Limits
|
||||
|
||||
When `max_cost_per_plan` is set, tools are blocked with a
|
||||
`ToolCostLimitExceededError` if `ToolExecutionContext.accumulated_cost`
|
||||
meets or exceeds the limit. Similarly, `max_total_cost` is checked against
|
||||
`ToolExecutionContext.total_accumulated_cost`. The caller is responsible
|
||||
for updating these cost fields after each tool execution.
|
||||
|
||||
### 8. Retry Limit
|
||||
|
||||
When `max_retries_per_step` is set, tools are blocked with a
|
||||
`ToolRetryLimitExceededError` if `ToolExecutionContext.step_retry_count`
|
||||
exceeds the limit. The caller is responsible for incrementing the retry
|
||||
count for each retry attempt.
|
||||
|
||||
## Error Hierarchy
|
||||
|
||||
```
|
||||
ToolRuntimeError
|
||||
|-- ToolAccessDeniedError (read-only plan violation)
|
||||
|-- ToolCheckpointRequiredError (checkpoint requirement violation)
|
||||
|-- ToolSafetyViolationError (unsafe tool or category violation)
|
||||
|-- ToolSandboxRequiredError (sandbox requirement violation)
|
||||
|-- ToolHumanApprovalRequiredError (human approval not granted)
|
||||
|-- ToolCostLimitExceededError (cost limit exceeded)
|
||||
|-- ToolRetryLimitExceededError (retry limit exceeded)
|
||||
|-- ToolNotActivatedError
|
||||
|-- ToolActivationError
|
||||
|-- ToolExecutionError
|
||||
+-- ToolDeactivationError
|
||||
```
|
||||
|
||||
## Backward Compatibility
|
||||
|
||||
When no `SafetyProfile` is attached to the `ToolExecutionContext` (i.e.,
|
||||
`safety_profile=None`), the safety-profile-specific checks (unsafe tool
|
||||
gating, skill category, sandbox requirement, human approval, cost limits,
|
||||
and retry limits) are skipped. Only the pre-existing `plan_read_only` and
|
||||
`require_checkpoints` context flags apply.
|
||||
|
||||
## Default Profile
|
||||
|
||||
The `DEFAULT_SAFETY_PROFILE` constant provides sensible defaults matching
|
||||
the specification:
|
||||
|
||||
```python
|
||||
DEFAULT_SAFETY_PROFILE = SafetyProfile(
|
||||
allowed_skill_categories=[],
|
||||
require_sandbox=True,
|
||||
require_checkpoints=True,
|
||||
require_human_approval=False,
|
||||
allow_unsafe_tools=False,
|
||||
max_cost_per_plan=None,
|
||||
max_retries_per_step=3,
|
||||
max_total_cost=None,
|
||||
)
|
||||
```
|
||||
|
||||
## Related
|
||||
|
||||
- [ADR-041: Safety Profile Extraction](../adr/ADR-041-safety-profile-extraction.md)
|
||||
- [Specification: Automation Profiles](../specification.md) -- Profile Precedence section
|
||||
- [Specification: Tool Capability Metadata](../specification.md) -- `unsafe` field
|
||||
@@ -131,12 +131,70 @@ Feature: Safety Profile Domain Model
|
||||
When I try to create a safety profile ref with empty name
|
||||
Then a safety validation error should be raised
|
||||
|
||||
# ---- resolve_safety_profile stub ----
|
||||
# ---- resolve_safety_profile precedence ----
|
||||
|
||||
Scenario: resolve_safety_profile raises NotImplementedError
|
||||
When I try to resolve safety profile
|
||||
Then a NotImplementedError should be raised
|
||||
And the resolve error should mention "not yet implemented"
|
||||
Scenario: resolve_safety_profile returns plan-level when all levels set
|
||||
Given a plan safety profile with allow_unsafe_tools true
|
||||
And an action safety profile with allow_unsafe_tools false
|
||||
And a project safety profile with allow_unsafe_tools false
|
||||
And a global safety profile with allow_unsafe_tools false
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "plan"
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
|
||||
Scenario: resolve_safety_profile returns action-level when no plan-level
|
||||
Given an action safety profile with allow_unsafe_tools true
|
||||
And a project safety profile with allow_unsafe_tools false
|
||||
And a global safety profile with allow_unsafe_tools false
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "action"
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
|
||||
Scenario: resolve_safety_profile returns project-level when no plan or action
|
||||
Given a project safety profile with allow_unsafe_tools true
|
||||
And a global safety profile with allow_unsafe_tools false
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "project"
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
|
||||
Scenario: resolve_safety_profile returns global-level when only global set
|
||||
Given a global safety profile with allow_unsafe_tools true
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "global"
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
|
||||
Scenario: resolve_safety_profile returns default when all levels are None
|
||||
When I resolve the safety profile with no levels
|
||||
Then the resolved provenance should be "global"
|
||||
And the resolved profile allow_unsafe_tools should be false
|
||||
And the resolved profile require_sandbox should be true
|
||||
|
||||
Scenario: resolve_safety_profile preserves all 8 fields from plan-level
|
||||
Given a full plan safety profile with all fields customized
|
||||
And a full action safety profile with defaults
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "plan"
|
||||
And the resolved profile require_sandbox should be false
|
||||
And the resolved profile require_checkpoints should be false
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
And the resolved profile require_human_approval should be true
|
||||
And the resolved profile max_cost_per_plan should be 50.0
|
||||
And the resolved profile max_retries_per_step should be 7
|
||||
And the resolved profile max_total_cost should be 200.0
|
||||
And the resolved profile allowed_skill_categories should be "code,test"
|
||||
|
||||
Scenario: resolve_safety_profile preserves all 8 fields from action-level
|
||||
Given a full action safety profile with all fields customized
|
||||
When I resolve the safety profile
|
||||
Then the resolved provenance should be "action"
|
||||
And the resolved profile require_sandbox should be false
|
||||
And the resolved profile require_checkpoints should be false
|
||||
And the resolved profile allow_unsafe_tools should be true
|
||||
And the resolved profile require_human_approval should be true
|
||||
And the resolved profile max_cost_per_plan should be 50.0
|
||||
And the resolved profile max_retries_per_step should be 7
|
||||
And the resolved profile max_total_cost should be 200.0
|
||||
And the resolved profile allowed_skill_categories should be "code,test"
|
||||
|
||||
# ---- model_dump round-trip ----
|
||||
|
||||
|
||||
@@ -0,0 +1,185 @@
|
||||
Feature: Safety Profile Enforcement
|
||||
As a developer
|
||||
I want the tool runtime to enforce safety profile constraints
|
||||
So that unsafe tools and disallowed skill categories are blocked
|
||||
|
||||
Background:
|
||||
Given a registered tool "test/writer" that writes and is safe
|
||||
And a registered tool "test/unsafe-tool" that is unsafe
|
||||
And a registered tool "test/reader" that is read-only and safe
|
||||
|
||||
# ---- Unsafe tool gating ----
|
||||
|
||||
Scenario: Unsafe tool is blocked when safety profile forbids unsafe tools
|
||||
Given a safety profile with allow_unsafe_tools false
|
||||
And a tool execution context with the safety profile
|
||||
When I enforce safety and try to run tool "test/unsafe-tool"
|
||||
Then a ToolSafetyViolationError should be raised
|
||||
And the safety violation error should mention "unsafe"
|
||||
|
||||
Scenario: Unsafe tool is allowed when safety profile permits unsafe tools
|
||||
Given a safety profile with allow_unsafe_tools true
|
||||
And a tool execution context with the safety profile
|
||||
When I enforce safety and run tool "test/unsafe-tool"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Safe tool is allowed regardless of allow_unsafe_tools setting
|
||||
Given a safety profile with allow_unsafe_tools false
|
||||
And a tool execution context with the safety profile
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Skill category enforcement ----
|
||||
|
||||
Scenario: Tool blocked when skill category not in allow-list
|
||||
Given a safety profile with allowed_skill_categories "code,test"
|
||||
And a tool execution context with the safety profile
|
||||
And the tool skill category is "deploy"
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolSafetyViolationError should be raised
|
||||
And the safety violation error should mention "deploy"
|
||||
And the safety violation error should mention "not in the allowed"
|
||||
|
||||
Scenario: Tool allowed when skill category is in allow-list
|
||||
Given a safety profile with allowed_skill_categories "code,test"
|
||||
And a tool execution context with the safety profile
|
||||
And the tool skill category is "code"
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: All categories allowed when allow-list is empty
|
||||
Given a safety profile with empty allowed_skill_categories
|
||||
And a tool execution context with the safety profile
|
||||
And the tool skill category is "anything"
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Checkpoint requirement from safety profile ----
|
||||
|
||||
Scenario: Non-checkpointable tool blocked when safety profile requires checkpoints
|
||||
Given a safety profile with require_checkpoints true
|
||||
And a tool execution context with the safety profile and require_checkpoints false
|
||||
When I enforce safety and try to run tool "test/writer"
|
||||
Then a safety ToolCheckpointRequiredError should be raised
|
||||
|
||||
Scenario: Non-checkpointable tool allowed when safety profile does not require checkpoints
|
||||
Given a safety profile with require_checkpoints false
|
||||
And a tool execution context with the safety profile and require_checkpoints false
|
||||
When I enforce safety and run tool "test/writer"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Context without safety profile (backward compatibility) ----
|
||||
|
||||
Scenario: Unsafe tool allowed when no safety profile on context
|
||||
Given a tool execution context without a safety profile
|
||||
When I enforce safety and run tool "test/unsafe-tool"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Enforcement still applies read-only checks without safety profile
|
||||
Given a tool execution context that is read-only without a safety profile
|
||||
When I enforce safety and try to run tool "test/writer"
|
||||
Then a safety ToolAccessDeniedError should be raised
|
||||
|
||||
# ---- Sandbox requirement ----
|
||||
|
||||
Scenario: Writing tool blocked when sandbox required but no sandbox_id
|
||||
Given a safety profile with require_sandbox true
|
||||
And a tool execution context with the safety profile and no sandbox_id
|
||||
When I enforce safety and try to run tool "test/writer"
|
||||
Then a ToolSandboxRequiredError should be raised
|
||||
And the sandbox error should mention "require_sandbox"
|
||||
|
||||
Scenario: Writing tool allowed when sandbox required and sandbox_id set
|
||||
Given a safety profile with require_sandbox true
|
||||
And a tool execution context with the safety profile and sandbox_id "sandbox-001"
|
||||
When I enforce safety and run tool "test/writer"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Writing tool allowed when sandbox not required and no sandbox_id
|
||||
Given a safety profile with require_sandbox false
|
||||
And a tool execution context with the safety profile and no sandbox_id
|
||||
When I enforce safety and run tool "test/writer"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Read-only tool allowed when sandbox required but no sandbox_id
|
||||
Given a safety profile with require_sandbox true
|
||||
And a tool execution context with the safety profile and no sandbox_id
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Human approval requirement ----
|
||||
|
||||
Scenario: Tool blocked when human approval required but not granted
|
||||
Given a safety profile with require_human_approval true
|
||||
And a tool execution context with the safety profile and no approval
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolHumanApprovalRequiredError should be raised
|
||||
And the approval error should mention "human approval"
|
||||
|
||||
Scenario: Tool allowed when human approval required and granted
|
||||
Given a safety profile with require_human_approval true
|
||||
And a tool execution context with the safety profile and human approval granted
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Tool allowed when human approval not required
|
||||
Given a safety profile with require_human_approval false
|
||||
And a tool execution context with the safety profile
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Cost limit enforcement ----
|
||||
|
||||
Scenario: Tool blocked when accumulated cost exceeds max_cost_per_plan
|
||||
Given a safety profile with max_cost_per_plan 10.0
|
||||
And a tool execution context with the safety profile and accumulated_cost 10.0
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolCostLimitExceededError should be raised
|
||||
And the cost error should mention "max_cost_per_plan"
|
||||
|
||||
Scenario: Tool allowed when accumulated cost is below max_cost_per_plan
|
||||
Given a safety profile with max_cost_per_plan 10.0
|
||||
And a tool execution context with the safety profile and accumulated_cost 5.0
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
Scenario: Tool blocked when total cost exceeds max_total_cost
|
||||
Given a safety profile with max_total_cost 100.0
|
||||
And a tool execution context with the safety profile and total_accumulated_cost 100.0
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolCostLimitExceededError should be raised
|
||||
And the cost error should mention "max_total_cost"
|
||||
|
||||
# ---- Retry limit enforcement ----
|
||||
|
||||
Scenario: Tool blocked when retry count exceeds max_retries_per_step
|
||||
Given a safety profile with max_retries_per_step 3
|
||||
And a tool execution context with the safety profile and step_retry_count 4
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolRetryLimitExceededError should be raised
|
||||
And the retry error should mention "max_retries_per_step"
|
||||
|
||||
Scenario: Tool allowed when retry count is within max_retries_per_step
|
||||
Given a safety profile with max_retries_per_step 3
|
||||
And a tool execution context with the safety profile and step_retry_count 2
|
||||
When I enforce safety and run tool "test/reader"
|
||||
Then the tool execution should succeed
|
||||
|
||||
# ---- Missing skill category metadata ----
|
||||
|
||||
Scenario: Tool blocked when skill category metadata missing and allow-list set
|
||||
Given a safety profile with allowed_skill_categories "code,test"
|
||||
And a tool execution context with the safety profile and no skill category metadata
|
||||
When I enforce safety and try to run tool "test/reader"
|
||||
Then a ToolSafetyViolationError should be raised
|
||||
And the safety violation error should mention "no skill category"
|
||||
|
||||
# ---- Combined constraints ----
|
||||
|
||||
Scenario: Multiple safety violations reported for first failing check
|
||||
Given a combined safety profile blocking unsafe tools with categories "code"
|
||||
And a tool execution context with the safety profile
|
||||
And the tool skill category is "deploy"
|
||||
When I enforce safety and try to run tool "test/unsafe-tool"
|
||||
Then a ToolSafetyViolationError should be raised
|
||||
And the safety violation error should mention "unsafe"
|
||||
@@ -0,0 +1,564 @@
|
||||
"""Step definitions for Safety Profile Enforcement tests.
|
||||
|
||||
Tests that ToolRuntime._enforce_capabilities correctly enforces safety
|
||||
profile constraints: unsafe tool gating, skill category allow-lists,
|
||||
checkpoint requirements, sandbox requirements, human approval,
|
||||
cost limits, and retry limits from the safety profile.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.domain.models.core.safety_profile import SafetyProfile
|
||||
from cleveragents.domain.models.core.tool import (
|
||||
Tool,
|
||||
ToolCapability,
|
||||
ToolSource,
|
||||
)
|
||||
from cleveragents.tool.context import ToolExecutionContext
|
||||
from cleveragents.tool.lifecycle import (
|
||||
ToolAccessDeniedError,
|
||||
ToolCheckpointRequiredError,
|
||||
ToolCostLimitExceededError,
|
||||
ToolDescriptor,
|
||||
ToolHumanApprovalRequiredError,
|
||||
ToolResult,
|
||||
ToolRetryLimitExceededError,
|
||||
ToolRuntime,
|
||||
ToolSafetyViolationError,
|
||||
ToolSandboxRequiredError,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers -- Stub ToolInstance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _StubToolInstance:
|
||||
"""Minimal ToolInstance for testing enforcement without real execution."""
|
||||
|
||||
def __init__(self, descriptor: ToolDescriptor) -> None:
|
||||
self._descriptor = descriptor
|
||||
|
||||
def discover(self) -> ToolDescriptor:
|
||||
return self._descriptor
|
||||
|
||||
def activate(self, ctx: ToolExecutionContext) -> None:
|
||||
pass
|
||||
|
||||
def execute(self, params: dict[str, Any], ctx: ToolExecutionContext) -> ToolResult:
|
||||
return ToolResult(success=True, data={"ok": True})
|
||||
|
||||
def deactivate(self, ctx: ToolExecutionContext) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _make_tool(
|
||||
name: str,
|
||||
*,
|
||||
writes: bool = False,
|
||||
read_only: bool = False,
|
||||
unsafe: bool = False,
|
||||
checkpointable: bool = False,
|
||||
) -> Tool:
|
||||
"""Create a Tool domain model with given capability flags."""
|
||||
return Tool(
|
||||
name=name,
|
||||
description=f"Test tool {name}",
|
||||
source=ToolSource.CUSTOM,
|
||||
code="pass",
|
||||
capability=ToolCapability(
|
||||
read_only=read_only,
|
||||
writes=writes,
|
||||
unsafe=unsafe,
|
||||
checkpointable=checkpointable,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Background: register tools
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('a registered tool "test/writer" that writes and is safe')
|
||||
def step_register_writer(context: Context) -> None:
|
||||
"""Register a tool that writes but is not unsafe."""
|
||||
context.enforcement_runtime = ToolRuntime()
|
||||
tool = _make_tool("test/writer", writes=True, unsafe=False)
|
||||
desc = ToolDescriptor(
|
||||
name=tool.name,
|
||||
description=tool.description,
|
||||
capability=tool.capability,
|
||||
)
|
||||
instance = _StubToolInstance(desc)
|
||||
context.enforcement_runtime.register_tool(tool, instance)
|
||||
|
||||
|
||||
@given('a registered tool "test/unsafe-tool" that is unsafe')
|
||||
def step_register_unsafe(context: Context) -> None:
|
||||
"""Register a tool marked as unsafe."""
|
||||
runtime: ToolRuntime = context.enforcement_runtime
|
||||
tool = _make_tool("test/unsafe-tool", writes=False, unsafe=True)
|
||||
desc = ToolDescriptor(
|
||||
name=tool.name,
|
||||
description=tool.description,
|
||||
capability=tool.capability,
|
||||
)
|
||||
instance = _StubToolInstance(desc)
|
||||
runtime.register_tool(tool, instance)
|
||||
|
||||
|
||||
@given('a registered tool "test/reader" that is read-only and safe')
|
||||
def step_register_reader(context: Context) -> None:
|
||||
"""Register a read-only safe tool."""
|
||||
runtime: ToolRuntime = context.enforcement_runtime
|
||||
tool = _make_tool("test/reader", read_only=True, unsafe=False)
|
||||
desc = ToolDescriptor(
|
||||
name=tool.name,
|
||||
description=tool.description,
|
||||
capability=tool.capability,
|
||||
)
|
||||
instance = _StubToolInstance(desc)
|
||||
runtime.register_tool(tool, instance)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Safety profile construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a safety profile with allow_unsafe_tools {val}")
|
||||
def step_safety_allow_unsafe(context: Context, val: str) -> None:
|
||||
"""Create a safety profile with specific allow_unsafe_tools value."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
allow_unsafe_tools=val.lower() == "true",
|
||||
require_checkpoints=False,
|
||||
require_sandbox=False,
|
||||
)
|
||||
|
||||
|
||||
@given('a safety profile with allowed_skill_categories "{cats}"')
|
||||
def step_safety_categories(context: Context, cats: str) -> None:
|
||||
"""Create a safety profile with specific allowed skill categories."""
|
||||
cat_list = [c.strip() for c in cats.split(",") if c.strip()]
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
allowed_skill_categories=cat_list,
|
||||
require_checkpoints=False,
|
||||
require_sandbox=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a safety profile with empty allowed_skill_categories")
|
||||
def step_safety_empty_categories(context: Context) -> None:
|
||||
"""Create a safety profile with empty allowed skill categories."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
allowed_skill_categories=[],
|
||||
require_checkpoints=False,
|
||||
require_sandbox=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a safety profile with require_checkpoints {val}")
|
||||
def step_safety_require_checkpoints(context: Context, val: str) -> None:
|
||||
"""Create a safety profile with specific require_checkpoints value."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
require_checkpoints=val.lower() == "true",
|
||||
require_sandbox=False,
|
||||
)
|
||||
|
||||
|
||||
@given('a combined safety profile blocking unsafe tools with categories "{cats}"')
|
||||
def step_safety_combined(context: Context, cats: str) -> None:
|
||||
"""Create a safety profile blocking unsafe tools with category constraints."""
|
||||
cat_list = [c.strip() for c in cats.split(",") if c.strip()]
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
allow_unsafe_tools=False,
|
||||
allowed_skill_categories=cat_list,
|
||||
require_checkpoints=False,
|
||||
require_sandbox=False,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool execution context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile")
|
||||
def step_ctx_with_safety(context: Context) -> None:
|
||||
"""Create a ToolExecutionContext with the current safety profile."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and require_checkpoints {val}")
|
||||
def step_ctx_with_safety_and_cp(context: Context, val: str) -> None:
|
||||
"""Create a context with safety profile and explicit require_checkpoints."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
require_checkpoints=val.lower() == "true",
|
||||
safety_profile=context.enforcement_safety,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context without a safety profile")
|
||||
def step_ctx_without_safety(context: Context) -> None:
|
||||
"""Create a ToolExecutionContext without a safety profile."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context that is read-only without a safety profile")
|
||||
def step_ctx_readonly(context: Context) -> None:
|
||||
"""Create a read-only ToolExecutionContext without a safety profile."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
plan_read_only=True,
|
||||
)
|
||||
|
||||
|
||||
@given('the tool skill category is "{category}"')
|
||||
def step_set_tool_category(context: Context, category: str) -> None:
|
||||
"""Set the tool_skill_category metadata on the execution context."""
|
||||
context.enforcement_ctx.metadata["tool_skill_category"] = category
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sandbox requirement context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a safety profile with require_sandbox {val}")
|
||||
def step_safety_require_sandbox(context: Context, val: str) -> None:
|
||||
"""Create a safety profile with specific require_sandbox value."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
require_sandbox=val.lower() == "true",
|
||||
require_checkpoints=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and no sandbox_id")
|
||||
def step_ctx_with_safety_no_sandbox(context: Context) -> None:
|
||||
"""Create a context with safety profile and no sandbox_id."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
sandbox_id=None,
|
||||
)
|
||||
|
||||
|
||||
@given('a tool execution context with the safety profile and sandbox_id "{sandbox_id}"')
|
||||
def step_ctx_with_safety_and_sandbox(context: Context, sandbox_id: str) -> None:
|
||||
"""Create a context with safety profile and a sandbox_id."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
sandbox_id=sandbox_id,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Human approval context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a safety profile with require_human_approval {val}")
|
||||
def step_safety_require_human_approval(context: Context, val: str) -> None:
|
||||
"""Create a safety profile with specific require_human_approval value."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
require_human_approval=val.lower() == "true",
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and no approval")
|
||||
def step_ctx_with_safety_no_approval(context: Context) -> None:
|
||||
"""Create a context with safety profile but no human approval."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and human approval granted")
|
||||
def step_ctx_with_safety_and_approval(context: Context) -> None:
|
||||
"""Create a context with safety profile and human approval granted."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
metadata={"human_approved": True},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cost limit context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a safety profile with max_cost_per_plan {val:g}")
|
||||
def step_safety_max_cost_per_plan(context: Context, val: float) -> None:
|
||||
"""Create a safety profile with max_cost_per_plan."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
max_cost_per_plan=val,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and accumulated_cost {val:g}")
|
||||
def step_ctx_with_safety_and_cost(context: Context, val: float) -> None:
|
||||
"""Create a context with safety profile and accumulated cost."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
accumulated_cost=val,
|
||||
)
|
||||
|
||||
|
||||
@given("a safety profile with max_total_cost {val:g}")
|
||||
def step_safety_max_total_cost(context: Context, val: float) -> None:
|
||||
"""Create a safety profile with max_total_cost."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
max_total_cost=val,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
|
||||
|
||||
@given(
|
||||
"a tool execution context with the safety profile"
|
||||
" and total_accumulated_cost {val:g}"
|
||||
)
|
||||
def step_ctx_with_safety_and_total_cost(context: Context, val: float) -> None:
|
||||
"""Create a context with safety profile and total accumulated cost."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
total_accumulated_cost=val,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Retry limit context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a safety profile with max_retries_per_step {val:d}")
|
||||
def step_safety_max_retries(context: Context, val: int) -> None:
|
||||
"""Create a safety profile with max_retries_per_step."""
|
||||
context.enforcement_safety = SafetyProfile(
|
||||
max_retries_per_step=val,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a tool execution context with the safety profile and step_retry_count {val:d}")
|
||||
def step_ctx_with_safety_and_retries(context: Context, val: int) -> None:
|
||||
"""Create a context with safety profile and step retry count."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
step_retry_count=val,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Missing metadata context construction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given(
|
||||
"a tool execution context with the safety profile and no skill category metadata"
|
||||
)
|
||||
def step_ctx_with_safety_no_category(context: Context) -> None:
|
||||
"""Create a context with safety profile but no tool_skill_category."""
|
||||
context.enforcement_ctx = ToolExecutionContext(
|
||||
plan_id="test-plan-001",
|
||||
safety_profile=context.enforcement_safety,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Execution steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when('I enforce safety and try to run tool "{tool_name}"')
|
||||
def step_try_execute(context: Context, tool_name: str) -> None:
|
||||
"""Try executing a tool through the safety-enforcing runtime, capturing errors."""
|
||||
runtime: ToolRuntime = context.enforcement_runtime
|
||||
ctx: ToolExecutionContext = context.enforcement_ctx
|
||||
context.enforcement_error = None
|
||||
context.enforcement_result = None
|
||||
try:
|
||||
result = runtime.execute(tool_name, {}, ctx)
|
||||
context.enforcement_result = result
|
||||
except (
|
||||
ToolSafetyViolationError,
|
||||
ToolAccessDeniedError,
|
||||
ToolCheckpointRequiredError,
|
||||
ToolSandboxRequiredError,
|
||||
ToolHumanApprovalRequiredError,
|
||||
ToolCostLimitExceededError,
|
||||
ToolRetryLimitExceededError,
|
||||
) as exc:
|
||||
context.enforcement_error = exc
|
||||
|
||||
|
||||
@when('I enforce safety and run tool "{tool_name}"')
|
||||
def step_execute(context: Context, tool_name: str) -> None:
|
||||
"""Execute a tool through the safety-enforcing runtime (expected to succeed)."""
|
||||
runtime: ToolRuntime = context.enforcement_runtime
|
||||
ctx: ToolExecutionContext = context.enforcement_ctx
|
||||
context.enforcement_result = runtime.execute(tool_name, {}, ctx)
|
||||
context.enforcement_error = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Assertions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("a ToolSafetyViolationError should be raised")
|
||||
def step_check_safety_violation(context: Context) -> None:
|
||||
"""Verify a ToolSafetyViolationError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolSafetyViolationError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolSafetyViolationError), (
|
||||
f"Expected ToolSafetyViolationError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("a safety ToolCheckpointRequiredError should be raised")
|
||||
def step_check_checkpoint_required(context: Context) -> None:
|
||||
"""Verify a ToolCheckpointRequiredError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolCheckpointRequiredError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolCheckpointRequiredError), (
|
||||
f"Expected ToolCheckpointRequiredError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("a safety ToolAccessDeniedError should be raised")
|
||||
def step_check_access_denied(context: Context) -> None:
|
||||
"""Verify a ToolAccessDeniedError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolAccessDeniedError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolAccessDeniedError), (
|
||||
f"Expected ToolAccessDeniedError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('the safety violation error should mention "{text}"')
|
||||
def step_check_violation_text(context: Context, text: str) -> None:
|
||||
"""Check that the safety violation error message contains the expected text."""
|
||||
error_str = str(context.enforcement_error)
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
@then("a ToolSandboxRequiredError should be raised")
|
||||
def step_check_sandbox_required(context: Context) -> None:
|
||||
"""Verify a ToolSandboxRequiredError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolSandboxRequiredError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolSandboxRequiredError), (
|
||||
f"Expected ToolSandboxRequiredError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('the sandbox error should mention "{text}"')
|
||||
def step_check_sandbox_error_text(context: Context, text: str) -> None:
|
||||
"""Check that the sandbox error message contains the expected text."""
|
||||
error_str = str(context.enforcement_error)
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
@then("a ToolHumanApprovalRequiredError should be raised")
|
||||
def step_check_human_approval_required(context: Context) -> None:
|
||||
"""Verify a ToolHumanApprovalRequiredError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolHumanApprovalRequiredError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolHumanApprovalRequiredError), (
|
||||
f"Expected ToolHumanApprovalRequiredError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('the approval error should mention "{text}"')
|
||||
def step_check_approval_error_text(context: Context, text: str) -> None:
|
||||
"""Check that the approval error message contains the expected text."""
|
||||
error_str = str(context.enforcement_error)
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
@then("a ToolCostLimitExceededError should be raised")
|
||||
def step_check_cost_limit_exceeded(context: Context) -> None:
|
||||
"""Verify a ToolCostLimitExceededError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolCostLimitExceededError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolCostLimitExceededError), (
|
||||
f"Expected ToolCostLimitExceededError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('the cost error should mention "{text}"')
|
||||
def step_check_cost_error_text(context: Context, text: str) -> None:
|
||||
"""Check that the cost error message contains the expected text."""
|
||||
error_str = str(context.enforcement_error)
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
@then("a ToolRetryLimitExceededError should be raised")
|
||||
def step_check_retry_limit_exceeded(context: Context) -> None:
|
||||
"""Verify a ToolRetryLimitExceededError was raised."""
|
||||
assert context.enforcement_error is not None, (
|
||||
"Expected ToolRetryLimitExceededError but no error was raised"
|
||||
)
|
||||
assert isinstance(context.enforcement_error, ToolRetryLimitExceededError), (
|
||||
f"Expected ToolRetryLimitExceededError, "
|
||||
f"got {type(context.enforcement_error).__name__}: "
|
||||
f"{context.enforcement_error}"
|
||||
)
|
||||
|
||||
|
||||
@then('the retry error should mention "{text}"')
|
||||
def step_check_retry_error_text(context: Context, text: str) -> None:
|
||||
"""Check that the retry error message contains the expected text."""
|
||||
error_str = str(context.enforcement_error)
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
@then("the tool execution should succeed")
|
||||
def step_check_execution_success(context: Context) -> None:
|
||||
"""Verify the tool execution succeeded."""
|
||||
assert context.enforcement_result is not None, "Expected a tool result but got None"
|
||||
assert context.enforcement_result.success is True, (
|
||||
f"Expected success=True, got: {context.enforcement_result}"
|
||||
)
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import then, when
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
from pydantic import ValidationError
|
||||
|
||||
@@ -230,18 +230,165 @@ def step_try_create_empty_ref(context: Context) -> None:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# resolve_safety_profile stub
|
||||
# resolve_safety_profile precedence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I try to resolve safety profile")
|
||||
def step_try_resolve(context: Context) -> None:
|
||||
"""Try calling resolve_safety_profile."""
|
||||
context.resolve_error = None
|
||||
try:
|
||||
resolve_safety_profile()
|
||||
except NotImplementedError as e:
|
||||
context.resolve_error = e
|
||||
@given("a plan safety profile with allow_unsafe_tools {val}")
|
||||
def step_given_plan_profile(context: Context, val: str) -> None:
|
||||
"""Set plan-level safety profile."""
|
||||
context.resolve_plan = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
||||
|
||||
|
||||
@given("an action safety profile with allow_unsafe_tools {val}")
|
||||
def step_given_action_profile(context: Context, val: str) -> None:
|
||||
"""Set action-level safety profile."""
|
||||
context.resolve_action = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
||||
|
||||
|
||||
@given("a project safety profile with allow_unsafe_tools {val}")
|
||||
def step_given_project_profile(context: Context, val: str) -> None:
|
||||
"""Set project-level safety profile."""
|
||||
context.resolve_project = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
||||
|
||||
|
||||
@given("a global safety profile with allow_unsafe_tools {val}")
|
||||
def step_given_global_profile(context: Context, val: str) -> None:
|
||||
"""Set global-level safety profile."""
|
||||
context.resolve_global = SafetyProfile(allow_unsafe_tools=val.lower() == "true")
|
||||
|
||||
|
||||
@given("a full plan safety profile with all fields customized")
|
||||
def step_given_full_plan_profile(context: Context) -> None:
|
||||
"""Set plan-level safety profile with all 8 fields customized."""
|
||||
context.resolve_plan = SafetyProfile(
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
allow_unsafe_tools=True,
|
||||
require_human_approval=True,
|
||||
allowed_skill_categories=["code", "test"],
|
||||
max_cost_per_plan=50.0,
|
||||
max_retries_per_step=7,
|
||||
max_total_cost=200.0,
|
||||
)
|
||||
|
||||
|
||||
@given("a full action safety profile with defaults")
|
||||
def step_given_full_action_defaults(context: Context) -> None:
|
||||
"""Set action-level safety profile with all defaults."""
|
||||
context.resolve_action = SafetyProfile()
|
||||
|
||||
|
||||
@given("a full action safety profile with all fields customized")
|
||||
def step_given_full_action_profile(context: Context) -> None:
|
||||
"""Set action-level safety profile with all 8 fields customized."""
|
||||
context.resolve_action = SafetyProfile(
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
allow_unsafe_tools=True,
|
||||
require_human_approval=True,
|
||||
allowed_skill_categories=["code", "test"],
|
||||
max_cost_per_plan=50.0,
|
||||
max_retries_per_step=7,
|
||||
max_total_cost=200.0,
|
||||
)
|
||||
|
||||
|
||||
@when("I resolve the safety profile")
|
||||
def step_resolve_profile(context: Context) -> None:
|
||||
"""Resolve safety profile with whatever levels are set."""
|
||||
plan = getattr(context, "resolve_plan", None)
|
||||
action = getattr(context, "resolve_action", None)
|
||||
project = getattr(context, "resolve_project", None)
|
||||
global_ = getattr(context, "resolve_global", None)
|
||||
profile, provenance = resolve_safety_profile(
|
||||
plan_profile=plan,
|
||||
action_profile=action,
|
||||
project_profile=project,
|
||||
global_profile=global_,
|
||||
)
|
||||
context.resolved_profile = profile
|
||||
context.resolved_provenance = provenance
|
||||
|
||||
|
||||
@when("I resolve the safety profile with no levels")
|
||||
def step_resolve_profile_none(context: Context) -> None:
|
||||
"""Resolve safety profile with all levels as None."""
|
||||
profile, provenance = resolve_safety_profile()
|
||||
context.resolved_profile = profile
|
||||
context.resolved_provenance = provenance
|
||||
|
||||
|
||||
@then('the resolved provenance should be "{expected}"')
|
||||
def step_check_resolved_provenance(context: Context, expected: str) -> None:
|
||||
"""Check resolved provenance value."""
|
||||
actual = context.resolved_provenance.value
|
||||
assert actual == expected, f"Expected provenance '{expected}', got '{actual}'"
|
||||
|
||||
|
||||
@then("the resolved profile allow_unsafe_tools should be {expected}")
|
||||
def step_check_resolved_allow_unsafe(context: Context, expected: str) -> None:
|
||||
"""Check resolved profile allow_unsafe_tools."""
|
||||
exp_bool = expected.lower() == "true"
|
||||
actual = context.resolved_profile.allow_unsafe_tools
|
||||
assert actual is exp_bool, f"Expected allow_unsafe_tools {exp_bool}, got {actual}"
|
||||
|
||||
|
||||
@then("the resolved profile require_sandbox should be {expected}")
|
||||
def step_check_resolved_require_sandbox(context: Context, expected: str) -> None:
|
||||
"""Check resolved profile require_sandbox."""
|
||||
exp_bool = expected.lower() == "true"
|
||||
actual = context.resolved_profile.require_sandbox
|
||||
assert actual is exp_bool, f"Expected require_sandbox {exp_bool}, got {actual}"
|
||||
|
||||
|
||||
@then("the resolved profile require_checkpoints should be {expected}")
|
||||
def step_check_resolved_require_checkpoints(context: Context, expected: str) -> None:
|
||||
"""Check resolved profile require_checkpoints."""
|
||||
exp_bool = expected.lower() == "true"
|
||||
actual = context.resolved_profile.require_checkpoints
|
||||
assert actual is exp_bool, f"Expected require_checkpoints {exp_bool}, got {actual}"
|
||||
|
||||
|
||||
@then("the resolved profile require_human_approval should be {expected}")
|
||||
def step_check_resolved_require_human_approval(context: Context, expected: str) -> None:
|
||||
"""Check resolved profile require_human_approval."""
|
||||
exp_bool = expected.lower() == "true"
|
||||
actual = context.resolved_profile.require_human_approval
|
||||
assert actual is exp_bool, (
|
||||
f"Expected require_human_approval {exp_bool}, got {actual}"
|
||||
)
|
||||
|
||||
|
||||
@then("the resolved profile max_cost_per_plan should be {expected:g}")
|
||||
def step_check_resolved_max_cost(context: Context, expected: float) -> None:
|
||||
"""Check resolved profile max_cost_per_plan."""
|
||||
actual = context.resolved_profile.max_cost_per_plan
|
||||
assert actual == expected, f"Expected max_cost_per_plan {expected}, got {actual}"
|
||||
|
||||
|
||||
@then("the resolved profile max_retries_per_step should be {expected:d}")
|
||||
def step_check_resolved_max_retries(context: Context, expected: int) -> None:
|
||||
"""Check resolved profile max_retries_per_step."""
|
||||
actual = context.resolved_profile.max_retries_per_step
|
||||
assert actual == expected, f"Expected max_retries_per_step {expected}, got {actual}"
|
||||
|
||||
|
||||
@then("the resolved profile max_total_cost should be {expected:g}")
|
||||
def step_check_resolved_max_total_cost(context: Context, expected: float) -> None:
|
||||
"""Check resolved profile max_total_cost."""
|
||||
actual = context.resolved_profile.max_total_cost
|
||||
assert actual == expected, f"Expected max_total_cost {expected}, got {actual}"
|
||||
|
||||
|
||||
@then('the resolved profile allowed_skill_categories should be "{expected}"')
|
||||
def step_check_resolved_categories(context: Context, expected: str) -> None:
|
||||
"""Check resolved profile allowed_skill_categories."""
|
||||
expected_cats = [c.strip() for c in expected.split(",") if c.strip()]
|
||||
actual = context.resolved_profile.allowed_skill_categories
|
||||
assert actual == expected_cats, (
|
||||
f"Expected allowed_skill_categories {expected_cats}, got {actual}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -469,6 +616,11 @@ def step_check_resolve_error_text(context: Context, text: str) -> None:
|
||||
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
|
||||
|
||||
|
||||
# NOTE: The above two steps are retained for backward compatibility.
|
||||
# They are no longer used by safety_profile.feature (the resolve stub was
|
||||
# replaced with real implementation tests) but may be referenced elsewhere.
|
||||
|
||||
|
||||
@then('the safety dump should have key "{key}"')
|
||||
def step_check_safety_dump_key(context: Context, key: str) -> None:
|
||||
"""Check dump has key."""
|
||||
|
||||
@@ -137,13 +137,16 @@ def _test_summary() -> None:
|
||||
print(f" Max total cost: {default.max_total_cost}")
|
||||
print(f" Categories: {default.allowed_skill_categories}")
|
||||
|
||||
# Verify resolve stub raises NotImplementedError
|
||||
try:
|
||||
resolve_safety_profile()
|
||||
print(" ERROR: resolve should raise NotImplementedError")
|
||||
sys.exit(1)
|
||||
except NotImplementedError:
|
||||
print(" resolve_safety_profile: NotImplementedError (expected)")
|
||||
# Verify resolve_safety_profile returns default with GLOBAL provenance
|
||||
# when all levels are None.
|
||||
resolved, provenance = resolve_safety_profile()
|
||||
assert resolved is DEFAULT_SAFETY_PROFILE, (
|
||||
f"Expected DEFAULT_SAFETY_PROFILE, got {resolved}"
|
||||
)
|
||||
assert provenance == SafetyProfileProvenance.GLOBAL, (
|
||||
f"Expected GLOBAL provenance, got {provenance}"
|
||||
)
|
||||
print(f" resolve_safety_profile: {provenance.value} (expected)")
|
||||
|
||||
print("summary-ok")
|
||||
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
"""Robot Framework helper for Safety Profile Enforcement smoke tests.
|
||||
|
||||
Provides a CLI-style interface for Robot to invoke safety profile
|
||||
resolution and enforcement operations. Exit code 0 = success, 1 = failure.
|
||||
|
||||
Usage:
|
||||
python robot/helper_safety_profile_enforcement.py resolve-precedence
|
||||
python robot/helper_safety_profile_enforcement.py resolve-default
|
||||
python robot/helper_safety_profile_enforcement.py enforce-unsafe-blocked
|
||||
python robot/helper_safety_profile_enforcement.py enforce-unsafe-allowed
|
||||
python robot/helper_safety_profile_enforcement.py enforce-category-blocked
|
||||
python robot/helper_safety_profile_enforcement.py enforce-category-allowed
|
||||
python robot/helper_safety_profile_enforcement.py enforce-checkpoint
|
||||
python robot/helper_safety_profile_enforcement.py enforce-sandbox-blocked
|
||||
python robot/helper_safety_profile_enforcement.py enforce-sandbox-allowed
|
||||
python robot/helper_safety_profile_enforcement.py enforce-human-approval
|
||||
python robot/helper_safety_profile_enforcement.py enforce-cost-limit
|
||||
python robot/helper_safety_profile_enforcement.py enforce-retry-limit
|
||||
python robot/helper_safety_profile_enforcement.py backward-compat
|
||||
python robot/helper_safety_profile_enforcement.py context-summary
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
# Ensure src is importable.
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
|
||||
from cleveragents.domain.models.core.safety_profile import ( # noqa: E402
|
||||
DEFAULT_SAFETY_PROFILE,
|
||||
SafetyProfile,
|
||||
SafetyProfileProvenance,
|
||||
resolve_safety_profile,
|
||||
)
|
||||
from cleveragents.domain.models.core.tool import ( # noqa: E402
|
||||
Tool,
|
||||
ToolCapability,
|
||||
ToolSource,
|
||||
)
|
||||
from cleveragents.tool.context import ToolExecutionContext # noqa: E402
|
||||
from cleveragents.tool.lifecycle import ( # noqa: E402
|
||||
ToolCheckpointRequiredError,
|
||||
ToolCostLimitExceededError,
|
||||
ToolDescriptor,
|
||||
ToolHumanApprovalRequiredError,
|
||||
ToolResult,
|
||||
ToolRetryLimitExceededError,
|
||||
ToolRuntime,
|
||||
ToolSafetyViolationError,
|
||||
ToolSandboxRequiredError,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stub tool instance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _StubInstance:
|
||||
"""Minimal ToolInstance for integration smoke tests."""
|
||||
|
||||
def __init__(self, name: str) -> None:
|
||||
self._name = name
|
||||
|
||||
def discover(self) -> ToolDescriptor:
|
||||
return ToolDescriptor(name=self._name, description=f"stub {self._name}")
|
||||
|
||||
def activate(self, ctx: ToolExecutionContext) -> None:
|
||||
pass
|
||||
|
||||
def execute(self, params: dict, ctx: ToolExecutionContext) -> ToolResult: # type: ignore[type-arg]
|
||||
return ToolResult(success=True, data={"ok": True})
|
||||
|
||||
def deactivate(self, ctx: ToolExecutionContext) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _make_tool(name: str, *, unsafe: bool = False, read_only: bool = False) -> Tool:
|
||||
return Tool(
|
||||
name=name,
|
||||
description=f"Test tool {name}",
|
||||
source=ToolSource.CUSTOM,
|
||||
code="pass",
|
||||
capability=ToolCapability(read_only=read_only, unsafe=unsafe),
|
||||
)
|
||||
|
||||
|
||||
def _make_runtime() -> ToolRuntime:
|
||||
rt = ToolRuntime()
|
||||
for name, unsafe, ro in [
|
||||
("test/safe", False, True),
|
||||
("test/unsafe", True, False),
|
||||
]:
|
||||
t = _make_tool(name, unsafe=unsafe, read_only=ro)
|
||||
rt.register_tool(t, _StubInstance(name))
|
||||
return rt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _cmd_resolve_precedence() -> int:
|
||||
"""Test that plan-level profile wins over all others."""
|
||||
plan = SafetyProfile(allow_unsafe_tools=True, require_sandbox=False)
|
||||
action = SafetyProfile(allow_unsafe_tools=False, require_sandbox=True)
|
||||
resolved, prov = resolve_safety_profile(plan_profile=plan, action_profile=action)
|
||||
if prov != SafetyProfileProvenance.PLAN:
|
||||
print(f"resolve-precedence-fail: expected PLAN, got {prov}")
|
||||
return 1
|
||||
if resolved.allow_unsafe_tools is not True:
|
||||
print("resolve-precedence-fail: allow_unsafe_tools should be True")
|
||||
return 1
|
||||
print("resolve-precedence-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_resolve_default() -> int:
|
||||
"""Test that all-None returns DEFAULT_SAFETY_PROFILE."""
|
||||
resolved, prov = resolve_safety_profile()
|
||||
if prov != SafetyProfileProvenance.GLOBAL:
|
||||
print(f"resolve-default-fail: expected GLOBAL, got {prov}")
|
||||
return 1
|
||||
if resolved is not DEFAULT_SAFETY_PROFILE:
|
||||
print("resolve-default-fail: not DEFAULT_SAFETY_PROFILE instance")
|
||||
return 1
|
||||
print("resolve-default-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_unsafe_blocked() -> int:
|
||||
"""Test that unsafe tool is blocked by safety profile."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
allow_unsafe_tools=False, require_sandbox=False, require_checkpoints=False
|
||||
)
|
||||
ctx = ToolExecutionContext(plan_id="robot-001", safety_profile=profile)
|
||||
try:
|
||||
rt.execute("test/unsafe", {}, ctx)
|
||||
print("enforce-unsafe-blocked-fail: no error raised")
|
||||
return 1
|
||||
except ToolSafetyViolationError:
|
||||
print("enforce-unsafe-blocked-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_unsafe_allowed() -> int:
|
||||
"""Test that unsafe tool runs when profile allows it."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
allow_unsafe_tools=True, require_sandbox=False, require_checkpoints=False
|
||||
)
|
||||
ctx = ToolExecutionContext(plan_id="robot-002", safety_profile=profile)
|
||||
result = rt.execute("test/unsafe", {}, ctx)
|
||||
if not result.success:
|
||||
print("enforce-unsafe-allowed-fail: execution not successful")
|
||||
return 1
|
||||
print("enforce-unsafe-allowed-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_category_blocked() -> int:
|
||||
"""Test that tool in wrong skill category is blocked."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
allowed_skill_categories=["code"],
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-003",
|
||||
safety_profile=profile,
|
||||
metadata={"tool_skill_category": "deploy"},
|
||||
)
|
||||
try:
|
||||
rt.execute("test/safe", {}, ctx)
|
||||
print("enforce-category-blocked-fail: no error raised")
|
||||
return 1
|
||||
except ToolSafetyViolationError:
|
||||
print("enforce-category-blocked-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_category_allowed() -> int:
|
||||
"""Test that tool in allowed skill category runs."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
allowed_skill_categories=["code"],
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-004",
|
||||
safety_profile=profile,
|
||||
metadata={"tool_skill_category": "code"},
|
||||
)
|
||||
result = rt.execute("test/safe", {}, ctx)
|
||||
if not result.success:
|
||||
print("enforce-category-allowed-fail: execution not successful")
|
||||
return 1
|
||||
print("enforce-category-allowed-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_checkpoint() -> int:
|
||||
"""Test that safety profile require_checkpoints blocks non-checkpointable tools."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(require_checkpoints=True, require_sandbox=False)
|
||||
ctx = ToolExecutionContext(plan_id="robot-005", safety_profile=profile)
|
||||
try:
|
||||
rt.execute("test/safe", {}, ctx)
|
||||
print("enforce-checkpoint-fail: no error raised")
|
||||
return 1
|
||||
except ToolCheckpointRequiredError:
|
||||
print("enforce-checkpoint-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_backward_compat() -> int:
|
||||
"""Test that unsafe tools run when no safety profile on context."""
|
||||
rt = _make_runtime()
|
||||
ctx = ToolExecutionContext(plan_id="robot-006")
|
||||
result = rt.execute("test/unsafe", {}, ctx)
|
||||
if not result.success:
|
||||
print("backward-compat-fail: execution not successful")
|
||||
return 1
|
||||
print("backward-compat-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_context_summary() -> int:
|
||||
"""Test that context summary includes safety profile flag."""
|
||||
profile = SafetyProfile(require_sandbox=False, require_checkpoints=False)
|
||||
ctx = ToolExecutionContext(plan_id="robot-007", safety_profile=profile)
|
||||
summary = ctx.as_summary()
|
||||
if summary.get("has_safety_profile") is not True:
|
||||
print(f"context-summary-fail: {summary}")
|
||||
return 1
|
||||
ctx2 = ToolExecutionContext(plan_id="robot-008")
|
||||
summary2 = ctx2.as_summary()
|
||||
if summary2.get("has_safety_profile") is not False:
|
||||
print(f"context-summary-fail: {summary2}")
|
||||
return 1
|
||||
print("context-summary-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_sandbox_blocked() -> int:
|
||||
"""Test that writing tool is blocked when sandbox required but missing."""
|
||||
rt = ToolRuntime()
|
||||
tool = Tool(
|
||||
name="test/writer",
|
||||
description="writes",
|
||||
source=ToolSource.CUSTOM,
|
||||
code="pass",
|
||||
capability=ToolCapability(writes=True),
|
||||
)
|
||||
rt.register_tool(tool, _StubInstance("test/writer"))
|
||||
profile = SafetyProfile(require_sandbox=True, require_checkpoints=False)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-009", safety_profile=profile, sandbox_id=None
|
||||
)
|
||||
try:
|
||||
rt.execute("test/writer", {}, ctx)
|
||||
print("enforce-sandbox-blocked-fail: no error raised")
|
||||
return 1
|
||||
except ToolSandboxRequiredError:
|
||||
print("enforce-sandbox-blocked-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_sandbox_allowed() -> int:
|
||||
"""Test that writing tool runs when sandbox is provided."""
|
||||
rt = ToolRuntime()
|
||||
tool = Tool(
|
||||
name="test/writer",
|
||||
description="writes",
|
||||
source=ToolSource.CUSTOM,
|
||||
code="pass",
|
||||
capability=ToolCapability(writes=True),
|
||||
)
|
||||
rt.register_tool(tool, _StubInstance("test/writer"))
|
||||
profile = SafetyProfile(require_sandbox=True, require_checkpoints=False)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-010",
|
||||
safety_profile=profile,
|
||||
sandbox_id="sandbox-001",
|
||||
)
|
||||
result = rt.execute("test/writer", {}, ctx)
|
||||
if not result.success:
|
||||
print("enforce-sandbox-allowed-fail: not successful")
|
||||
return 1
|
||||
print("enforce-sandbox-allowed-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_human_approval() -> int:
|
||||
"""Test that tools are blocked when human approval is required."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
require_human_approval=True,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
ctx = ToolExecutionContext(plan_id="robot-011", safety_profile=profile)
|
||||
try:
|
||||
rt.execute("test/safe", {}, ctx)
|
||||
print("enforce-human-approval-fail: no error raised")
|
||||
return 1
|
||||
except ToolHumanApprovalRequiredError:
|
||||
print("enforce-human-approval-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_cost_limit() -> int:
|
||||
"""Test that tools are blocked when cost limit is exceeded."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
max_cost_per_plan=10.0,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-012",
|
||||
safety_profile=profile,
|
||||
accumulated_cost=10.0,
|
||||
)
|
||||
try:
|
||||
rt.execute("test/safe", {}, ctx)
|
||||
print("enforce-cost-limit-fail: no error raised")
|
||||
return 1
|
||||
except ToolCostLimitExceededError:
|
||||
print("enforce-cost-limit-ok")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_enforce_retry_limit() -> int:
|
||||
"""Test that tools are blocked when retry limit is exceeded."""
|
||||
rt = _make_runtime()
|
||||
profile = SafetyProfile(
|
||||
max_retries_per_step=3,
|
||||
require_sandbox=False,
|
||||
require_checkpoints=False,
|
||||
)
|
||||
ctx = ToolExecutionContext(
|
||||
plan_id="robot-013",
|
||||
safety_profile=profile,
|
||||
step_retry_count=4,
|
||||
)
|
||||
try:
|
||||
rt.execute("test/safe", {}, ctx)
|
||||
print("enforce-retry-limit-fail: no error raised")
|
||||
return 1
|
||||
except ToolRetryLimitExceededError:
|
||||
print("enforce-retry-limit-ok")
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_COMMANDS: dict[str, Callable[[], int]] = {
|
||||
"resolve-precedence": _cmd_resolve_precedence,
|
||||
"resolve-default": _cmd_resolve_default,
|
||||
"enforce-unsafe-blocked": _cmd_enforce_unsafe_blocked,
|
||||
"enforce-unsafe-allowed": _cmd_enforce_unsafe_allowed,
|
||||
"enforce-category-blocked": _cmd_enforce_category_blocked,
|
||||
"enforce-category-allowed": _cmd_enforce_category_allowed,
|
||||
"enforce-checkpoint": _cmd_enforce_checkpoint,
|
||||
"enforce-sandbox-blocked": _cmd_enforce_sandbox_blocked,
|
||||
"enforce-sandbox-allowed": _cmd_enforce_sandbox_allowed,
|
||||
"enforce-human-approval": _cmd_enforce_human_approval,
|
||||
"enforce-cost-limit": _cmd_enforce_cost_limit,
|
||||
"enforce-retry-limit": _cmd_enforce_retry_limit,
|
||||
"backward-compat": _cmd_backward_compat,
|
||||
"context-summary": _cmd_context_summary,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Entry point called by Robot Framework ``Run Process``."""
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: helper_safety_profile_enforcement.py <command>")
|
||||
return 1
|
||||
command = sys.argv[1]
|
||||
handler = _COMMANDS.get(command)
|
||||
if handler is None:
|
||||
print(f"Unknown command: {command}")
|
||||
return 1
|
||||
return handler()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,124 @@
|
||||
*** Settings ***
|
||||
Documentation Safety Profile Enforcement integration smoke tests
|
||||
... Verifies resolve_safety_profile precedence and
|
||||
... ToolRuntime safety enforcement (unsafe tool gating,
|
||||
... skill category allow-lists, checkpoint requirements).
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Variables ***
|
||||
${HELPER} ${CURDIR}/helper_safety_profile_enforcement.py
|
||||
|
||||
*** Test Cases ***
|
||||
Resolve Safety Profile Precedence Plan Over Action
|
||||
[Documentation] Plan-level profile takes precedence over action-level
|
||||
${result}= Run Process ${PYTHON} ${HELPER} resolve-precedence cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} resolve-precedence-ok
|
||||
|
||||
Resolve Safety Profile Default When All None
|
||||
[Documentation] All-None resolution returns DEFAULT_SAFETY_PROFILE with GLOBAL provenance
|
||||
${result}= Run Process ${PYTHON} ${HELPER} resolve-default cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} resolve-default-ok
|
||||
|
||||
Unsafe Tool Blocked By Safety Profile
|
||||
[Documentation] Unsafe tool raises ToolSafetyViolationError when profile forbids
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-unsafe-blocked cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-unsafe-blocked-ok
|
||||
|
||||
Unsafe Tool Allowed By Safety Profile
|
||||
[Documentation] Unsafe tool executes when profile permits allow_unsafe_tools
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-unsafe-allowed cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-unsafe-allowed-ok
|
||||
|
||||
Skill Category Blocked By Safety Profile
|
||||
[Documentation] Tool in wrong skill category is blocked by allow-list
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-category-blocked cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-category-blocked-ok
|
||||
|
||||
Skill Category Allowed By Safety Profile
|
||||
[Documentation] Tool in correct skill category runs normally
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-category-allowed cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-category-allowed-ok
|
||||
|
||||
Checkpoint Required By Safety Profile
|
||||
[Documentation] Non-checkpointable tool blocked when profile requires checkpoints
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-checkpoint cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-checkpoint-ok
|
||||
|
||||
Backward Compatibility Without Safety Profile
|
||||
[Documentation] Unsafe tool runs when no safety profile on context
|
||||
${result}= Run Process ${PYTHON} ${HELPER} backward-compat cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} backward-compat-ok
|
||||
|
||||
Sandbox Required Blocks Writing Tool Without Sandbox Id
|
||||
[Documentation] Writing tool raises ToolSandboxRequiredError when sandbox required but missing
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-sandbox-blocked cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-sandbox-blocked-ok
|
||||
|
||||
Sandbox Required Allows Writing Tool With Sandbox Id
|
||||
[Documentation] Writing tool executes when sandbox required and sandbox_id set
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-sandbox-allowed cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-sandbox-allowed-ok
|
||||
|
||||
Human Approval Required Blocks Tool Without Approval
|
||||
[Documentation] Tool blocked when require_human_approval=True and no approval
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-human-approval cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-human-approval-ok
|
||||
|
||||
Cost Limit Blocks Tool When Exceeded
|
||||
[Documentation] Tool blocked when accumulated cost exceeds max_cost_per_plan
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-cost-limit cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-cost-limit-ok
|
||||
|
||||
Retry Limit Blocks Tool When Exceeded
|
||||
[Documentation] Tool blocked when step retry count exceeds max_retries_per_step
|
||||
${result}= Run Process ${PYTHON} ${HELPER} enforce-retry-limit cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} enforce-retry-limit-ok
|
||||
|
||||
Context Summary Includes Safety Profile Flag
|
||||
[Documentation] ToolExecutionContext.as_summary includes has_safety_profile
|
||||
${result}= Run Process ${PYTHON} ${HELPER} context-summary cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-summary-ok
|
||||
@@ -20,8 +20,10 @@ while the SafetyProfile enforces hard safety constraints.
|
||||
|
||||
## Enforcement
|
||||
|
||||
Local-mode enforcement is a stub that raises ``NotImplementedError``.
|
||||
Future server-mode releases will implement real enforcement.
|
||||
Safety profile enforcement is implemented in the tool execution pipeline.
|
||||
The ``resolve_safety_profile`` function selects the highest-precedence
|
||||
profile, and ``ToolRuntime._enforce_capabilities`` enforces the resolved
|
||||
profile's constraints at tool activation and execution time.
|
||||
|
||||
Based on ``docs/specification.md`` sections "Automation Profiles" and
|
||||
"Guardrails".
|
||||
@@ -280,7 +282,7 @@ DEFAULT_SAFETY_PROFILE: SafetyProfile = SafetyProfile(
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resolution stub
|
||||
# Resolution
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -296,30 +298,33 @@ def resolve_safety_profile(
|
||||
Resolution order (highest to lowest):
|
||||
``plan > action > project > global``.
|
||||
|
||||
In local mode, enforcement is not implemented — this stub selects
|
||||
the highest-precedence profile but raises ``NotImplementedError``
|
||||
when called, signalling that real enforcement is deferred.
|
||||
The effective profile is determined at ``plan use`` time. Once
|
||||
resolved, the profile is **locked to that plan** -- subsequent
|
||||
changes to project or global profiles do not affect running plans.
|
||||
|
||||
When all levels are ``None`` the ``DEFAULT_SAFETY_PROFILE`` is
|
||||
returned with ``GLOBAL`` provenance.
|
||||
|
||||
Args:
|
||||
plan_profile: Profile set directly on the plan.
|
||||
plan_profile: Profile set directly on the plan (highest).
|
||||
action_profile: Profile inherited from the action template.
|
||||
project_profile: Profile inherited from the project.
|
||||
global_profile: Global default profile.
|
||||
global_profile: Global default profile (lowest).
|
||||
|
||||
Returns:
|
||||
Tuple of (resolved profile, provenance).
|
||||
|
||||
Raises:
|
||||
NotImplementedError: Always, in local mode.
|
||||
Tuple of ``(resolved_profile, provenance)`` where *provenance*
|
||||
indicates which level supplied the winning profile.
|
||||
"""
|
||||
candidates: list[tuple[SafetyProfile | None, SafetyProfileProvenance]] = [
|
||||
(plan_profile, SafetyProfileProvenance.PLAN),
|
||||
(action_profile, SafetyProfileProvenance.ACTION),
|
||||
(project_profile, SafetyProfileProvenance.PROJECT),
|
||||
(global_profile, SafetyProfileProvenance.GLOBAL),
|
||||
]
|
||||
|
||||
# Future implementation should resolve in this order:
|
||||
# 1. plan_profile (highest precedence)
|
||||
# 2. action_profile
|
||||
# 3. project_profile
|
||||
# 4. global_profile (lowest precedence)
|
||||
# Return the first non-None profile with its provenance.
|
||||
raise NotImplementedError(
|
||||
"Safety profile enforcement is not yet implemented in local mode. "
|
||||
"Resolution precedence: plan > action > project > global."
|
||||
)
|
||||
for profile, provenance in candidates:
|
||||
if profile is not None:
|
||||
return profile, provenance
|
||||
|
||||
# No explicit profile at any level -- fall back to the default.
|
||||
return DEFAULT_SAFETY_PROFILE, SafetyProfileProvenance.GLOBAL
|
||||
|
||||
@@ -30,14 +30,19 @@ from cleveragents.tool.lifecycle import (
|
||||
ToolAccessDeniedError,
|
||||
ToolActivationError,
|
||||
ToolCheckpointRequiredError,
|
||||
ToolCostLimitExceededError,
|
||||
ToolDeactivationError,
|
||||
ToolDescriptor,
|
||||
ToolExecutionError,
|
||||
ToolHumanApprovalRequiredError,
|
||||
ToolInstance,
|
||||
ToolLifecycleCache,
|
||||
ToolNotActivatedError,
|
||||
ToolRetryLimitExceededError,
|
||||
ToolRuntime,
|
||||
ToolRuntimeError,
|
||||
ToolSafetyViolationError,
|
||||
ToolSandboxRequiredError,
|
||||
)
|
||||
from cleveragents.tool.lifecycle import (
|
||||
ToolResult as LifecycleToolResult,
|
||||
@@ -90,20 +95,25 @@ __all__ = [
|
||||
"ToolCallingRuntime",
|
||||
"ToolCancelledError",
|
||||
"ToolCheckpointRequiredError",
|
||||
"ToolCostLimitExceededError",
|
||||
"ToolDeactivationError",
|
||||
"ToolDescriptor",
|
||||
"ToolError",
|
||||
"ToolExecutionContext",
|
||||
"ToolExecutionError",
|
||||
"ToolExecutionTrace",
|
||||
"ToolHumanApprovalRequiredError",
|
||||
"ToolInstance",
|
||||
"ToolLifecycleCache",
|
||||
"ToolNotActivatedError",
|
||||
"ToolRegistry",
|
||||
"ToolResult",
|
||||
"ToolRetryLimitExceededError",
|
||||
"ToolRunner",
|
||||
"ToolRuntime",
|
||||
"ToolRuntimeError",
|
||||
"ToolSafetyViolationError",
|
||||
"ToolSandboxRequiredError",
|
||||
"ToolSchemaValidationError",
|
||||
"ToolSpec",
|
||||
"classify_tool_error",
|
||||
|
||||
@@ -18,6 +18,8 @@ from typing import Any
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from cleveragents.domain.models.core.safety_profile import SafetyProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bound resource placeholder
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -165,6 +167,13 @@ class ToolExecutionContext:
|
||||
Whether the plan is read-only (restricts tool access).
|
||||
require_checkpoints:
|
||||
Whether the plan requires all tools to be checkpointable.
|
||||
safety_profile:
|
||||
Optional resolved safety profile for the plan. When provided,
|
||||
the tool runtime enforces constraints such as
|
||||
``allow_unsafe_tools``, ``allowed_skill_categories``,
|
||||
``require_checkpoints``, ``require_sandbox``,
|
||||
``require_human_approval``, cost limits, and retry limits
|
||||
from this profile.
|
||||
sandbox_id:
|
||||
Optional sandbox identifier for the execution.
|
||||
resources:
|
||||
@@ -172,7 +181,20 @@ class ToolExecutionContext:
|
||||
cancellation_token:
|
||||
Token for cancelling long-running tools.
|
||||
metadata:
|
||||
Additional plan/execution metadata.
|
||||
Additional plan/execution metadata. The key
|
||||
``human_approved`` (bool) is checked when the safety profile
|
||||
has ``require_human_approval=True``.
|
||||
accumulated_cost:
|
||||
Running total of cost (USD) for the current plan. Updated
|
||||
by the caller after each tool execution. Compared against
|
||||
the safety profile's ``max_cost_per_plan``.
|
||||
total_accumulated_cost:
|
||||
Running total of cost (USD) across all plans. Updated by
|
||||
the caller. Compared against ``max_total_cost``.
|
||||
step_retry_count:
|
||||
Number of retry attempts for the current action step.
|
||||
Updated by the caller. Compared against
|
||||
``max_retries_per_step``.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -181,18 +203,26 @@ class ToolExecutionContext:
|
||||
plan_id: str,
|
||||
plan_read_only: bool = False,
|
||||
require_checkpoints: bool = False,
|
||||
safety_profile: SafetyProfile | None = None,
|
||||
sandbox_id: str | None = None,
|
||||
resources: dict[str, BoundResource] | None = None,
|
||||
cancellation_token: CancellationToken | None = None,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
accumulated_cost: float = 0.0,
|
||||
total_accumulated_cost: float = 0.0,
|
||||
step_retry_count: int = 0,
|
||||
) -> None:
|
||||
self.plan_id = plan_id
|
||||
self.plan_read_only = plan_read_only
|
||||
self.require_checkpoints = require_checkpoints
|
||||
self.safety_profile = safety_profile
|
||||
self.sandbox_id = sandbox_id
|
||||
self.resources: dict[str, BoundResource] = resources or {}
|
||||
self.cancellation_token = cancellation_token or CancellationToken()
|
||||
self.metadata: dict[str, Any] = metadata or {}
|
||||
self.accumulated_cost = accumulated_cost
|
||||
self.total_accumulated_cost = total_accumulated_cost
|
||||
self.step_retry_count = step_retry_count
|
||||
self.changes: list[Change] = []
|
||||
self.traces: list[ToolExecutionTrace] = []
|
||||
|
||||
@@ -222,8 +252,12 @@ class ToolExecutionContext:
|
||||
result["plan_id"] = self.plan_id
|
||||
result["plan_read_only"] = self.plan_read_only
|
||||
result["require_checkpoints"] = self.require_checkpoints
|
||||
result["has_safety_profile"] = self.safety_profile is not None
|
||||
result["sandbox_id"] = self.sandbox_id
|
||||
result["resource_count"] = len(self.resources)
|
||||
result["change_count"] = len(self.changes)
|
||||
result["trace_count"] = len(self.traces)
|
||||
result["accumulated_cost"] = self.accumulated_cost
|
||||
result["total_accumulated_cost"] = self.total_accumulated_cost
|
||||
result["step_retry_count"] = self.step_retry_count
|
||||
return result
|
||||
|
||||
@@ -31,6 +31,12 @@ ToolRuntime
|
||||
|------------|-----------|-------|
|
||||
| Read-only plan | ``writes=True`` | ``ToolAccessDeniedError`` |
|
||||
| Checkpoint required | ``checkpointable=False`` | ``ToolCheckpointRequiredError`` |
|
||||
| Unsafe tool blocked | ``unsafe=True`` + forbids | ``ToolSafetyViolationError`` |
|
||||
| Skill category denied | category not in allow-list | ``ToolSafetyViolationError`` |
|
||||
| Sandbox required | ``writes=True`` + no sandbox | ``ToolSandboxRequiredError`` |
|
||||
| Human approval req. | not approved | ``ToolHumanApprovalRequiredError`` |
|
||||
| Cost limit exceeded | accumulated cost >= limit | ``ToolCostLimitExceededError`` |
|
||||
| Retry limit exceeded | retries > max | ``ToolRetryLimitExceededError`` |
|
||||
|
||||
## Error Hierarchy
|
||||
|
||||
@@ -38,6 +44,11 @@ ToolRuntime
|
||||
ToolRuntimeError
|
||||
|-- ToolAccessDeniedError
|
||||
|-- ToolCheckpointRequiredError
|
||||
|-- ToolSafetyViolationError
|
||||
|-- ToolSandboxRequiredError
|
||||
|-- ToolHumanApprovalRequiredError
|
||||
|-- ToolCostLimitExceededError
|
||||
|-- ToolRetryLimitExceededError
|
||||
|-- ToolNotActivatedError
|
||||
|-- ToolActivationError
|
||||
|-- ToolExecutionError
|
||||
@@ -112,6 +123,49 @@ class ToolExecutionError(ToolRuntimeError):
|
||||
"""Raised when tool execution fails."""
|
||||
|
||||
|
||||
class ToolSafetyViolationError(ToolRuntimeError):
|
||||
"""Raised when a tool is invoked in violation of safety profile constraints.
|
||||
|
||||
For example, an unsafe tool invoked when the safety profile has
|
||||
``allow_unsafe_tools=False``, or a tool whose skill category is not
|
||||
in the profile's ``allowed_skill_categories``.
|
||||
"""
|
||||
|
||||
|
||||
class ToolSandboxRequiredError(ToolRuntimeError):
|
||||
"""Raised when sandbox isolation is required but not available.
|
||||
|
||||
When the safety profile has ``require_sandbox=True`` and a tool with
|
||||
``writes=True`` is invoked without a ``sandbox_id`` on the execution
|
||||
context, this error is raised to prevent unprotected writes.
|
||||
"""
|
||||
|
||||
|
||||
class ToolHumanApprovalRequiredError(ToolRuntimeError):
|
||||
"""Raised when human approval is required but not granted.
|
||||
|
||||
When the safety profile has ``require_human_approval=True`` and the
|
||||
execution context does not carry ``human_approved=True`` in its
|
||||
metadata, this error is raised to block execution until approval.
|
||||
"""
|
||||
|
||||
|
||||
class ToolCostLimitExceededError(ToolRuntimeError):
|
||||
"""Raised when a cost limit from the safety profile is exceeded.
|
||||
|
||||
Covers both ``max_cost_per_plan`` (per-plan accumulated cost) and
|
||||
``max_total_cost`` (cross-plan accumulated cost).
|
||||
"""
|
||||
|
||||
|
||||
class ToolRetryLimitExceededError(ToolRuntimeError):
|
||||
"""Raised when the retry limit per step is exceeded.
|
||||
|
||||
When the safety profile's ``max_retries_per_step`` is reached for a
|
||||
given action step, further tool executions for that step are blocked.
|
||||
"""
|
||||
|
||||
|
||||
class ToolDeactivationError(ToolRuntimeError):
|
||||
"""Raised when tool deactivation fails."""
|
||||
|
||||
@@ -480,13 +534,10 @@ class ToolRuntime:
|
||||
|
||||
tool = self._get_tool(tool_name)
|
||||
|
||||
# 2. Enforce capability flags
|
||||
self._enforce_capabilities(tool, ctx)
|
||||
|
||||
# 3. Auto-activate if needed
|
||||
# 2. Auto-activate if needed (activate already enforces capabilities)
|
||||
self.activate(tool_name, ctx)
|
||||
|
||||
# 4. Validate inputs
|
||||
# 3. Validate inputs
|
||||
if tool.input_schema:
|
||||
try:
|
||||
validate_tool_input(params, tool.input_schema)
|
||||
@@ -496,7 +547,7 @@ class ToolRuntime:
|
||||
error=f"Input validation failed: {exc}",
|
||||
)
|
||||
|
||||
# 5. Execute with tracing
|
||||
# 4. Execute with tracing
|
||||
started_at = datetime.now(UTC)
|
||||
trace = ToolExecutionTrace(
|
||||
tool_name=tool_name,
|
||||
@@ -553,7 +604,7 @@ class ToolRuntime:
|
||||
|
||||
ctx.add_trace(trace)
|
||||
|
||||
# 6. Validate outputs
|
||||
# 5. Validate outputs
|
||||
if tool.output_schema and result.data is not None:
|
||||
try:
|
||||
validate_tool_output(result.data, tool.output_schema)
|
||||
@@ -663,6 +714,29 @@ class ToolRuntime:
|
||||
def _enforce_capabilities(tool: Tool, ctx: ToolExecutionContext) -> None:
|
||||
"""Enforce capability constraints against the execution context.
|
||||
|
||||
Checks are evaluated in this order:
|
||||
|
||||
1. **Read-only plan** -- tools with ``writes=True`` are blocked.
|
||||
2. **Checkpoint requirement** -- non-checkpointable tools blocked
|
||||
when the plan or safety profile requires checkpoints.
|
||||
3. **Unsafe tool gating** -- tools with ``unsafe=True`` are blocked
|
||||
when the safety profile has ``allow_unsafe_tools=False``.
|
||||
4. **Skill category allow-list** -- tools whose skill category
|
||||
(carried in ``ctx.metadata["tool_skill_category"]``) is not in
|
||||
the profile's ``allowed_skill_categories`` are blocked. An
|
||||
empty allow-list means all categories are permitted.
|
||||
5. **Sandbox requirement** -- tools with ``writes=True`` are blocked
|
||||
when the safety profile has ``require_sandbox=True`` and no
|
||||
``sandbox_id`` is set on the context.
|
||||
6. **Human approval** -- all tools are blocked when the safety
|
||||
profile has ``require_human_approval=True`` and the context
|
||||
metadata does not carry ``human_approved=True``.
|
||||
7. **Cost limits** -- tools are blocked when the accumulated cost
|
||||
on the context exceeds the safety profile's
|
||||
``max_cost_per_plan`` or ``max_total_cost``.
|
||||
8. **Retry limit** -- tools are blocked when the step retry count
|
||||
on the context exceeds ``max_retries_per_step``.
|
||||
|
||||
Raises
|
||||
------
|
||||
ToolAccessDeniedError:
|
||||
@@ -670,22 +744,126 @@ class ToolRuntime:
|
||||
ToolCheckpointRequiredError:
|
||||
If the plan requires checkpoints and the tool is not
|
||||
``checkpointable``.
|
||||
ToolSafetyViolationError:
|
||||
If the tool violates the safety profile (unsafe tool gating
|
||||
or skill category restriction).
|
||||
ToolSandboxRequiredError:
|
||||
If sandbox is required but no sandbox_id is set and the tool
|
||||
writes.
|
||||
ToolHumanApprovalRequiredError:
|
||||
If human approval is required but not granted.
|
||||
ToolCostLimitExceededError:
|
||||
If accumulated cost exceeds the plan or total cost limit.
|
||||
ToolRetryLimitExceededError:
|
||||
If the step retry count exceeds max_retries_per_step.
|
||||
"""
|
||||
cap = tool.capability
|
||||
|
||||
# Read-only plan cannot use tools that write.
|
||||
# Any tool with writes=True is blocked regardless of its read_only flag
|
||||
# (the ToolCapability model validator already prevents read_only=True
|
||||
# combined with writes=True, but we enforce defensively).
|
||||
# 1. Read-only plan cannot use tools that write.
|
||||
if ctx.plan_read_only and cap.writes:
|
||||
raise ToolAccessDeniedError(
|
||||
f"Tool '{tool.name}' has writes=True but plan "
|
||||
f"'{ctx.plan_id}' is read-only"
|
||||
)
|
||||
|
||||
# Checkpoint-required plan cannot use non-checkpointable tools
|
||||
if ctx.require_checkpoints and not cap.checkpointable:
|
||||
# 2. Checkpoint-required plan cannot use non-checkpointable tools.
|
||||
require_cp = ctx.require_checkpoints
|
||||
if ctx.safety_profile is not None:
|
||||
require_cp = require_cp or ctx.safety_profile.require_checkpoints
|
||||
if require_cp and not cap.checkpointable:
|
||||
raise ToolCheckpointRequiredError(
|
||||
f"Tool '{tool.name}' is not checkpointable but plan "
|
||||
f"'{ctx.plan_id}' requires checkpoints"
|
||||
)
|
||||
|
||||
# 3. Unsafe tool gating via safety profile.
|
||||
if (
|
||||
ctx.safety_profile is not None
|
||||
and cap.unsafe
|
||||
and not ctx.safety_profile.allow_unsafe_tools
|
||||
):
|
||||
raise ToolSafetyViolationError(
|
||||
f"Tool '{tool.name}' is marked unsafe but the safety "
|
||||
f"profile forbids unsafe tools "
|
||||
f"(allow_unsafe_tools=False)"
|
||||
)
|
||||
|
||||
# 4. Skill category allow-list via safety profile.
|
||||
if ctx.safety_profile is not None:
|
||||
allowed = ctx.safety_profile.allowed_skill_categories
|
||||
if allowed:
|
||||
tool_category = ctx.metadata.get("tool_skill_category")
|
||||
if tool_category is None:
|
||||
raise ToolSafetyViolationError(
|
||||
f"Tool '{tool.name}' has no skill category "
|
||||
f"metadata ('tool_skill_category') but the "
|
||||
f"safety profile restricts categories "
|
||||
f"to: {allowed}"
|
||||
)
|
||||
if tool_category not in allowed:
|
||||
raise ToolSafetyViolationError(
|
||||
f"Tool '{tool.name}' belongs to skill category "
|
||||
f"'{tool_category}' which is not in the allowed "
|
||||
f"categories: {allowed}"
|
||||
)
|
||||
|
||||
# 5. Sandbox requirement via safety profile.
|
||||
if (
|
||||
ctx.safety_profile is not None
|
||||
and ctx.safety_profile.require_sandbox
|
||||
and cap.writes
|
||||
and ctx.sandbox_id is None
|
||||
):
|
||||
raise ToolSandboxRequiredError(
|
||||
f"Tool '{tool.name}' has writes=True but the safety "
|
||||
f"profile requires sandbox isolation "
|
||||
f"(require_sandbox=True) and no sandbox_id is set "
|
||||
f"on the execution context"
|
||||
)
|
||||
|
||||
# 6. Human approval requirement via safety profile.
|
||||
if (
|
||||
ctx.safety_profile is not None
|
||||
and ctx.safety_profile.require_human_approval
|
||||
and not ctx.metadata.get("human_approved", False)
|
||||
):
|
||||
raise ToolHumanApprovalRequiredError(
|
||||
f"Tool '{tool.name}' requires human approval before "
|
||||
f"execution (require_human_approval=True) but no "
|
||||
f"approval has been recorded in the context metadata"
|
||||
)
|
||||
|
||||
# 7. Cost limits via safety profile.
|
||||
if ctx.safety_profile is not None:
|
||||
if (
|
||||
ctx.safety_profile.max_cost_per_plan is not None
|
||||
and ctx.accumulated_cost >= ctx.safety_profile.max_cost_per_plan
|
||||
):
|
||||
raise ToolCostLimitExceededError(
|
||||
f"Plan '{ctx.plan_id}' accumulated cost "
|
||||
f"({ctx.accumulated_cost}) has reached the "
|
||||
f"max_cost_per_plan limit "
|
||||
f"({ctx.safety_profile.max_cost_per_plan})"
|
||||
)
|
||||
if (
|
||||
ctx.safety_profile.max_total_cost is not None
|
||||
and ctx.total_accumulated_cost >= ctx.safety_profile.max_total_cost
|
||||
):
|
||||
raise ToolCostLimitExceededError(
|
||||
f"Total accumulated cost "
|
||||
f"({ctx.total_accumulated_cost}) has reached the "
|
||||
f"max_total_cost limit "
|
||||
f"({ctx.safety_profile.max_total_cost})"
|
||||
)
|
||||
|
||||
# 8. Retry limit via safety profile.
|
||||
if (
|
||||
ctx.safety_profile is not None
|
||||
and ctx.step_retry_count > ctx.safety_profile.max_retries_per_step
|
||||
):
|
||||
raise ToolRetryLimitExceededError(
|
||||
f"Step retry count ({ctx.step_retry_count}) exceeds "
|
||||
f"max_retries_per_step "
|
||||
f"({ctx.safety_profile.max_retries_per_step}) for plan "
|
||||
f"'{ctx.plan_id}'"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user