diff --git a/benchmarks/safety_profile_bench.py b/benchmarks/safety_profile_bench.py new file mode 100644 index 000000000..6e198e7ff --- /dev/null +++ b/benchmarks/safety_profile_bench.py @@ -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) diff --git a/docs/reference/safety_profiles.md b/docs/reference/safety_profiles.md new file mode 100644 index 000000000..eb54534f4 --- /dev/null +++ b/docs/reference/safety_profiles.md @@ -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 diff --git a/features/safety_profile.feature b/features/safety_profile.feature index d52b783de..3729cc894 100644 --- a/features/safety_profile.feature +++ b/features/safety_profile.feature @@ -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 ---- diff --git a/features/safety_profile_enforcement.feature b/features/safety_profile_enforcement.feature new file mode 100644 index 000000000..262a7b713 --- /dev/null +++ b/features/safety_profile_enforcement.feature @@ -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" diff --git a/features/steps/safety_profile_enforcement_steps.py b/features/steps/safety_profile_enforcement_steps.py new file mode 100644 index 000000000..5a7946ab9 --- /dev/null +++ b/features/steps/safety_profile_enforcement_steps.py @@ -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}" + ) diff --git a/features/steps/safety_profile_steps.py b/features/steps/safety_profile_steps.py index 546fb7971..67f3be37f 100644 --- a/features/steps/safety_profile_steps.py +++ b/features/steps/safety_profile_steps.py @@ -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.""" diff --git a/robot/helper_safety_profile.py b/robot/helper_safety_profile.py index 76bbefdfa..ec1efc6ff 100644 --- a/robot/helper_safety_profile.py +++ b/robot/helper_safety_profile.py @@ -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") diff --git a/robot/helper_safety_profile_enforcement.py b/robot/helper_safety_profile_enforcement.py new file mode 100644 index 000000000..4a4e00c5a --- /dev/null +++ b/robot/helper_safety_profile_enforcement.py @@ -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 ") + 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()) diff --git a/robot/safety_profile_enforcement.robot b/robot/safety_profile_enforcement.robot new file mode 100644 index 000000000..87856b4bc --- /dev/null +++ b/robot/safety_profile_enforcement.robot @@ -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 diff --git a/src/cleveragents/domain/models/core/safety_profile.py b/src/cleveragents/domain/models/core/safety_profile.py index 58dc5dcf7..090f55160 100644 --- a/src/cleveragents/domain/models/core/safety_profile.py +++ b/src/cleveragents/domain/models/core/safety_profile.py @@ -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 diff --git a/src/cleveragents/tool/__init__.py b/src/cleveragents/tool/__init__.py index 42b5199ad..5e5395d0b 100644 --- a/src/cleveragents/tool/__init__.py +++ b/src/cleveragents/tool/__init__.py @@ -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", diff --git a/src/cleveragents/tool/context.py b/src/cleveragents/tool/context.py index ebc5bc290..ed293db8a 100644 --- a/src/cleveragents/tool/context.py +++ b/src/cleveragents/tool/context.py @@ -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 diff --git a/src/cleveragents/tool/lifecycle.py b/src/cleveragents/tool/lifecycle.py index 27f0d802d..5c4b4a6b1 100644 --- a/src/cleveragents/tool/lifecycle.py +++ b/src/cleveragents/tool/lifecycle.py @@ -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}'" + )