atlas/home-voice-agent/safety/confirmations/risk.py

"""
Risk classification for tool calls.

Categorizes tools by risk level and determines if confirmation is required.
"""

from enum import Enum
from typing import Dict, Set, Optional


class RiskLevel(Enum):
    """Risk levels for tool calls."""
    LOW = "low"  # No confirmation needed
    MEDIUM = "medium"  # Optional confirmation
    HIGH = "high"  # Confirmation required
    CRITICAL = "critical"  # Explicit confirmation required


class RiskClassifier:
    """Classifies tool calls by risk level."""
    
    def __init__(self):
        """Initialize risk classifier with tool risk mappings."""
        # High-risk tools that require confirmation
        self.high_risk_tools: Set[str] = {
            "send_email",
            "create_calendar_event",
            "update_calendar_event",
            "delete_calendar_event",
            "set_smart_home_scene",
            "toggle_smart_device",
            "adjust_thermostat",
            "delete_note",
            "delete_task",
        }
        
        # Medium-risk tools (optional confirmation)
        self.medium_risk_tools: Set[str] = {
            "update_task_status",  # Moving tasks between columns
            "append_to_note",  # Modifying existing notes
            "create_reminder",  # Creating reminders
        }
        
        # Low-risk tools (no confirmation needed)
        self.low_risk_tools: Set[str] = {
            "get_current_time",
            "get_date",
            "get_timezone_info",
            "convert_timezone",
            "get_weather",
            "list_timers",
            "list_tasks",
            "list_notes",
            "read_note",
            "search_notes",
            "create_timer",  # Timers are low-risk
            "create_note",  # Creating new notes is low-risk
            "add_task",  # Adding tasks is low-risk
        }
        
        # Critical-risk tools (explicit confirmation required)
        self.critical_risk_tools: Set[str] = {
            "send_email",  # Sending emails is critical
            "delete_calendar_event",  # Deleting calendar events
            "set_smart_home_scene",  # Smart home control
        }
    
    def classify_risk(self, tool_name: str, **kwargs) -> RiskLevel:
        """
        Classify risk level for a tool call.
        
        Args:
            tool_name: Name of the tool
            **kwargs: Tool-specific parameters that might affect risk
            
        Returns:
            RiskLevel enum
        """
        # Check critical first
        if tool_name in self.critical_risk_tools:
            return RiskLevel.CRITICAL
        
        # Check high risk
        if tool_name in self.high_risk_tools:
            return RiskLevel.HIGH
        
        # Check medium risk
        if tool_name in self.medium_risk_tools:
            # Can be elevated to HIGH based on context
            if self._is_high_risk_context(tool_name, **kwargs):
                return RiskLevel.HIGH
            return RiskLevel.MEDIUM
        
        # Default to low risk
        if tool_name in self.low_risk_tools:
            return RiskLevel.LOW
        
        # Unknown tool - default to medium risk (safe default)
        return RiskLevel.MEDIUM
    
    def _is_high_risk_context(self, tool_name: str, **kwargs) -> bool:
        """Check if context elevates risk level."""
        # Example: Updating task to "done" might be higher risk
        if tool_name == "update_task_status":
            new_status = kwargs.get("status", "").lower()
            if new_status in ["done", "cancelled"]:
                return True
        
        # Example: Appending to important notes
        if tool_name == "append_to_note":
            note_path = kwargs.get("note_path", "")
            if "important" in note_path.lower() or "critical" in note_path.lower():
                return True
        
        return False
    
    def requires_confirmation(self, tool_name: str, **kwargs) -> bool:
        """
        Check if tool call requires confirmation.
        
        Args:
            tool_name: Name of the tool
            **kwargs: Tool-specific parameters
            
        Returns:
            True if confirmation is required
        """
        risk = self.classify_risk(tool_name, **kwargs)
        return risk in [RiskLevel.HIGH, RiskLevel.CRITICAL]
    
    def get_confirmation_message(self, tool_name: str, **kwargs) -> str:
        """
        Generate confirmation message for tool call.
        
        Args:
            tool_name: Name of the tool
            **kwargs: Tool-specific parameters
            
        Returns:
            Confirmation message
        """
        risk = self.classify_risk(tool_name, **kwargs)
        
        if risk == RiskLevel.CRITICAL:
            return f"⚠️ CRITICAL ACTION: I'm about to {self._describe_action(tool_name, **kwargs)}. This cannot be undone. Do you want to proceed? (Yes/No)"
        elif risk == RiskLevel.HIGH:
            return f"⚠️ I'm about to {self._describe_action(tool_name, **kwargs)}. Do you want to proceed? (Yes/No)"
        else:
            return f"I'm about to {self._describe_action(tool_name, **kwargs)}. Proceed? (Yes/No)"
    
    def _describe_action(self, tool_name: str, **kwargs) -> str:
        """Generate human-readable description of action."""
        descriptions = {
            "send_email": f"send an email to {kwargs.get('to', 'recipient')}",
            "create_calendar_event": f"create a calendar event: {kwargs.get('title', 'event')}",
            "update_calendar_event": f"update calendar event: {kwargs.get('event_id', 'event')}",
            "delete_calendar_event": f"delete calendar event: {kwargs.get('event_id', 'event')}",
            "set_smart_home_scene": f"set smart home scene: {kwargs.get('scene_name', 'scene')}",
            "toggle_smart_device": f"toggle device: {kwargs.get('device_name', 'device')}",
            "adjust_thermostat": f"adjust thermostat to {kwargs.get('temperature', 'temperature')}°",
            "delete_note": f"delete note: {kwargs.get('note_path', 'note')}",
            "delete_task": f"delete task: {kwargs.get('task_path', 'task')}",
            "update_task_status": f"update task status to {kwargs.get('status', 'status')}",
            "append_to_note": f"append to note: {kwargs.get('note_path', 'note')}",
            "create_reminder": f"create a reminder: {kwargs.get('message', 'reminder')}",
        }
        
        return descriptions.get(tool_name, f"execute {tool_name}")


# Global classifier instance
_classifier = RiskClassifier()


def get_classifier() -> RiskClassifier:
    """Get the global risk classifier instance."""
    return _classifier