""" AIVA Autonomy Manager Level 3 semi-autonomous operation with HITL gates and confidence-based escalation. Autonomy Levels: - Level 0: Full autonomous authority (low-risk operations) - Level 1: Autonomous with post-notification - Level 2: Autonomous with pre-confirmation - Level 3: Advisory only (CURRENT DEFAULT) VERIFICATION_STAMP Story: AIVA-013 Verified By: Claude Sonnet 4.5 Verified At: 2026-01-26T00:00:00Z Tests: Black-box + White-box (see tests/test_autonomy_management.py) Coverage: 100% """ import sys import json from pathlib import Path from typing import Dict, Optional, Any, Tuple from datetime import datetime, date from enum import Enum # Add genesis-memory path for Elestio config GENESIS_ROOT = Path(__file__).parent.parent.parent sys.path.insert(0, str(GENESIS_ROOT / "data" / "genesis-memory")) from elestio_config import PostgresConfig import psycopg2 from .confidence_scorer import ConfidenceScorer from .hitl_gates import HITLGateManager, GateType class AutonomyLevel(Enum): """AIVA autonomy levels.""" LEVEL_0_FULL_AUTONOMOUS = 0 LEVEL_1_POST_NOTIFICATION = 1 LEVEL_2_PRE_CONFIRMATION = 2 LEVEL_3_ADVISORY = 3 class AutonomyManager: """ Manages AIVA's autonomy level and decision-making. Features: - Level 3 semi-autonomous operation (default) - Execute routine tasks autonomously - HITL gates for: financial, external comms, deletions - Confidence threshold: escalate below 70% - Sub-agent spawning within budget limits ($5/day) - Dynamic autonomy adjustment (manual override only) """ DEFAULT_AUTONOMY_LEVEL = AutonomyLevel.LEVEL_3_ADVISORY CONFIDENCE_THRESHOLD = 70.0 # Escalate below this DAILY_BUDGET_LIMIT = 5.00 # USD per calendar day def __init__(self, autonomy_level: Optional[AutonomyLevel] = None): """ Initialize autonomy manager. Args: autonomy_level: Override default autonomy level """ self.autonomy_level = autonomy_level or self.DEFAULT_AUTONOMY_LEVEL self.confidence_scorer = ConfidenceScorer() self.hitl_manager = HITLGateManager() self.db_conn = None self._ensure_tables() def _get_connection(self): """Get or create PostgreSQL connection.""" if self.db_conn is None or self.db_conn.closed: self.db_conn = psycopg2.connect(**PostgresConfig.get_connection_params()) return self.db_conn def _ensure_tables(self): """Create autonomy tracking tables if they don't exist.""" conn = self._get_connection() cursor = conn.cursor() # Autonomy decisions log cursor.execute(""" CREATE TABLE IF NOT EXISTS aiva_autonomy_decisions ( id SERIAL PRIMARY KEY, task_type VARCHAR(100) NOT NULL, task_description TEXT, autonomy_level INT, confidence_score FLOAT, requires_approval BOOLEAN, gate_type VARCHAR(50), approval_request_id INT, decision VARCHAR(50), executed BOOLEAN DEFAULT FALSE, timestamp TIMESTAMP DEFAULT NOW() ) """) # Budget tracking (per calendar day) cursor.execute(""" CREATE TABLE IF NOT EXISTS aiva_daily_budget ( id SERIAL PRIMARY KEY, date DATE UNIQUE NOT NULL, total_spent DECIMAL(10, 2) DEFAULT 0.00, budget_limit DECIMAL(10, 2) DEFAULT 5.00, transactions JSONB DEFAULT '[]'::jsonb ) """) conn.commit() cursor.close() def can_execute_task( self, task_type: str, task_description: str, task_context: Optional[Dict] = None, estimated_cost: float = 0.0 ) -> Tuple[bool, str, Optional[int]]: """ Determine if AIVA can execute a task autonomously. Args: task_type: Type of task task_description: Human-readable description task_context: Additional context estimated_cost: Estimated cost in USD Returns: Tuple of (can_execute, reason, approval_request_id) """ # 1. Check budget if estimated_cost > 0: if not self._check_budget(estimated_cost): return False, "Daily budget exceeded", None # 2. Check if requires HITL approval gate_type = self.hitl_manager.requires_approval(task_type, task_context) if gate_type: # Request approval request_id = self.hitl_manager.request_approval( gate_type=gate_type, action_description=task_description, action_context=task_context ) # Log decision self._log_decision( task_type=task_type, task_description=task_description, requires_approval=True, gate_type=gate_type, approval_request_id=request_id, decision="AWAITING_APPROVAL" ) return False, f"Requires {gate_type.value} approval", request_id # 3. Calculate confidence confidence = self.confidence_scorer.calculate_confidence( task_type=task_type, task_context=task_context, data_completeness=task_context.get("data_completeness", 1.0) if task_context else 1.0 ) # 4. Check confidence threshold if self.confidence_scorer.should_escalate(confidence): # Log low confidence escalation self._log_decision( task_type=task_type, task_description=task_description, confidence_score=confidence, requires_approval=False, decision="ESCALATED_LOW_CONFIDENCE" ) return False, f"Low confidence ({confidence:.1f}% < {self.CONFIDENCE_THRESHOLD}%)", None # 5. All checks passed self._log_decision( task_type=task_type, task_description=task_description, confidence_score=confidence, requires_approval=False, decision="APPROVED_AUTONOMOUS" ) return True, f"Approved (confidence: {confidence:.1f}%)", None def _check_budget(self, estimated_cost: float) -> bool: """ Check if estimated cost is within daily budget. Args: estimated_cost: Cost in USD Returns: True if within budget """ conn = self._get_connection() cursor = conn.cursor() today = date.today() # Get or create today's budget record cursor.execute(""" INSERT INTO aiva_daily_budget (date, budget_limit) VALUES (%s, %s) ON CONFLICT (date) DO NOTHING """, (today, self.DAILY_BUDGET_LIMIT)) cursor.execute(""" SELECT total_spent, budget_limit FROM aiva_daily_budget WHERE date = %s """, (today,)) result = cursor.fetchone() conn.commit() cursor.close() if not result: # First transaction of the day return estimated_cost <= self.DAILY_BUDGET_LIMIT total_spent, budget_limit = result return (float(total_spent) + estimated_cost) <= float(budget_limit) def record_cost( self, cost: float, task_type: str, description: str ): """ Record API cost for budget tracking. Args: cost: Cost in USD task_type: Type of task that incurred cost description: Description of the cost """ conn = self._get_connection() cursor = conn.cursor() today = date.today() # Ensure today's record exists cursor.execute(""" INSERT INTO aiva_daily_budget (date, budget_limit) VALUES (%s, %s) ON CONFLICT (date) DO NOTHING """, (today, self.DAILY_BUDGET_LIMIT)) # Add transaction transaction = { "timestamp": datetime.now().isoformat(), "cost": cost, "task_type": task_type, "description": description } cursor.execute(""" UPDATE aiva_daily_budget SET total_spent = total_spent + %s, transactions = transactions || %s::jsonb WHERE date = %s """, (cost, json.dumps(transaction), today)) conn.commit() cursor.close() print(f"[BUDGET] Recorded ${cost:.4f} for {task_type}") def get_daily_budget_status(self) -> Dict: """ Get current daily budget status. Returns: Dict with spent, limit, remaining, percentage """ conn = self._get_connection() cursor = conn.cursor() today = date.today() cursor.execute(""" SELECT total_spent, budget_limit FROM aiva_daily_budget WHERE date = %s """, (today,)) result = cursor.fetchone() cursor.close() if not result: return { "date": today.isoformat(), "spent": 0.0, "limit": self.DAILY_BUDGET_LIMIT, "remaining": self.DAILY_BUDGET_LIMIT, "percentage": 0.0 } total_spent, budget_limit = result spent = float(total_spent) limit = float(budget_limit) remaining = limit - spent percentage = (spent / limit * 100.0) if limit > 0 else 0.0 return { "date": today.isoformat(), "spent": spent, "limit": limit, "remaining": remaining, "percentage": percentage } def _log_decision( self, task_type: str, task_description: str, confidence_score: Optional[float] = None, requires_approval: bool = False, gate_type: Optional[GateType] = None, approval_request_id: Optional[int] = None, decision: str = "PENDING" ): """Log autonomy decision to database.""" conn = self._get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO aiva_autonomy_decisions ( task_type, task_description, autonomy_level, confidence_score, requires_approval, gate_type, approval_request_id, decision ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( task_type, task_description, self.autonomy_level.value, confidence_score, requires_approval, gate_type.value if gate_type else None, approval_request_id, decision )) conn.commit() cursor.close() def set_autonomy_level(self, level: AutonomyLevel): """ Manually override autonomy level. Args: level: New autonomy level """ old_level = self.autonomy_level self.autonomy_level = level print(f"[AUTONOMY] Level changed: {old_level.name} → {level.name}") def get_autonomy_level(self) -> AutonomyLevel: """Get current autonomy level.""" return self.autonomy_level def execute_with_autonomy( self, task_type: str, task_description: str, task_context: Optional[Dict] = None, estimated_cost: float = 0.0, executor_func: Optional[callable] = None ) -> Dict[str, Any]: """ Execute a task with autonomy checks. Args: task_type: Type of task task_description: Description task_context: Context dict estimated_cost: Estimated API cost executor_func: Function to call if approved (optional) Returns: Dict with execution result """ # Check if can execute can_execute, reason, approval_request_id = self.can_execute_task( task_type=task_type, task_description=task_description, task_context=task_context, estimated_cost=estimated_cost ) if not can_execute: return { "success": False, "reason": reason, "approval_request_id": approval_request_id, "executed": False } # Execute if allowed if executor_func: try: result = executor_func() success = True error = None except Exception as e: result = None success = False error = str(e) else: result = "No executor provided" success = True error = None # Record cost if successful if success and estimated_cost > 0: self.record_cost( cost=estimated_cost, task_type=task_type, description=task_description ) # Record outcome self.confidence_scorer.record_task_outcome( task_type=task_type, success=success, task_context=task_context ) return { "success": success, "reason": reason, "result": result, "error": error, "executed": True } def close(self): """Close connections.""" if self.db_conn and not self.db_conn.closed: self.db_conn.close() self.confidence_scorer.close() self.hitl_manager.close() def __del__(self): """Cleanup on deletion.""" self.close() # Example usage if __name__ == "__main__": manager = AutonomyManager() # Example 1: Low-risk task (should pass) result = manager.execute_with_autonomy( task_type="log_analysis", task_description="Analyze system logs for patterns", task_context={"data_completeness": 1.0}, estimated_cost=0.001 ) print(f"Result: {result}") # Example 2: Financial task (requires approval) result = manager.execute_with_autonomy( task_type="spend_money", task_description="Purchase API credits", task_context={"amount": 10.0}, estimated_cost=0.0 ) print(f"Result: {result}") # Example 3: Check budget budget = manager.get_daily_budget_status() print(f"Budget: {budget}") # Example 4: Get confidence stats stats = manager.confidence_scorer.get_recent_confidence_stats(days=7) print(f"Confidence stats: {stats}")