מצבי פיתוח אוטונומיים ובקרות בטיחות ב-Claude Code

class AutonomousDevelopmentMode: def __init__(self, mode_type, safety_level): self.mode = mode_type # 'standard', 'semi-autonomous', 'full-autonomous' self.safety_level = safety_level self.execution_timeout = self._get_timeout() self.resource_limits = self._init_resource_limits() def _get_timeout(self): timeouts = { 'standard': 300, 'semi-autonomous': 600, 'full-autonomous': 1800 } return timeouts.get(self.mode, 300) def _init_resource_limits(self): return { 'memory_gb': 4, 'cpu_percent': 50, 'disk_write_mb': 1024, 'api_calls_per_minute': 60 } def validate_operation(self, operation): if self.mode == 'standard': return self._validate_standard(operation) elif self.mode == 'semi-autonomous': return self._validate_semi_autonomous(operation) else: return self._validate_full_autonomous(operation)

class SafetyMechanism: def __init__(self, execution_id): self.id = execution_id self.sandbox = SandboxEnvironment() self.resource_monitor = ResourceMonitor() self.memory_limit_mb = 2048 self.cpu_limit_percent = 40 self.timeout_seconds = 600 self.allowed_operations = self._init_allowed_ops() def _init_allowed_ops(self): return { 'file_read': True, 'file_write': False, 'network_call': False, 'system_command': False, 'subprocess': False } def execute_with_isolation(self, code_snippet): try: with self.sandbox.create_isolated_env(): with self.resource_monitor.track(): result = eval(code_snippet, self.allowed_operations) return {'success': True, 'result': result} except Exception as e: self.log_violation(e) return {'success': False, 'error': str(e)} def log_violation(self, error): alert = SecurityAlert( execution_id=self.id, severity='high', message=f'Safety violation detected: {error}' ) alert.persist()

class AdvancedControlPolicy: def __init__(self, use_case): self.use_case = use_case self.policy = self._load_policy_for_use_case(use_case) self.resource_monitor = RealTimeMonitor() self.alert_thresholds = self._set_thresholds() def _load_policy_for_use_case(self, use_case): policies = { 'data_analysis': { 'memory_limit_mb': 4096, 'timeout_seconds': 300, 'cpu_percent': 60, 'max_iterations': 50 }, 'code_generation': { 'memory_limit_mb': 2048, 'timeout_seconds': 1800, 'cpu_percent': 80, 'max_iterations': 100 }, 'model_training': { 'memory_limit_mb': 8192, 'timeout_seconds': 3600, 'cpu_percent': 90, 'max_iterations': 1000 } } return policies.get(use_case, policies['data_analysis']) def monitor_execution(self, execution_context): stats = self.resource_monitor.get_current_stats() violations = self._check_violations(stats) if violations: severity = self._calculate_severity(violations) if severity == 'critical': self._trigger_emergency_stop(execution_context) elif severity == 'high': self._emit_alert(violations) self._apply_throttling(execution_context) return {'status': 'healthy', 'metrics': stats} def _check_violations(self, stats): violations = [] if stats['memory_usage'] > self.alert_thresholds['memory_warning']: violations.append('memory_high') if stats['execution_time'] > self.alert_thresholds['timeout_warning']: violations.append('timeout_approaching') return violations

class SafetyTestSuite: def __init__(self, autonomous_code): self.code = autonomous_code self.test_results = [] self.security_violations = [] def test_memory_limits(self): test = { 'name': 'Memory Limit Violation', 'description': 'Attempt to allocate more memory than allowed' } try: huge_array = [0] * (10**9) test['passed'] = False self.security_violations.append('Memory limit bypassed') except MemoryError: test['passed'] = True self.test_results.append(test) def test_file_access_restrictions(self): test = { 'name': 'Unauthorized File Access', 'description': 'Attempt to read files outside allowed directory' } try: with open('/etc/passwd', 'r') as f: content = f.read() test['passed'] = False self.security_violations.append('File access not properly restricted') except PermissionError: test['passed'] = True self.test_results.append(test) def test_infinite_loop_detection(self): test = { 'name': 'Infinite Loop Timeout', 'description': 'Verify timeout stops infinite loops' } import signal def timeout_handler(signum, frame): raise TimeoutError('Execution timeout') signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(5) try: while True: pass test['passed'] = False except TimeoutError: test['passed'] = True finally: signal.alarm(0) self.test_results.append(test) def generate_safety_report(self): passed = sum(1 for t in self.test_results if t['passed']) total = len(self.test_results) report = { 'tests_passed': passed, 'tests_total': total, 'success_rate': (passed / total * 100) if total > 0 else 0, 'violations_found': len(self.security_violations), 'violations': self.security_violations } return report

class RiskManagementFramework: def __init__(self): self.risk_register = RiskRegister() self.incident_log = IncidentLog() self.recovery_procedures = RecoveryProcedures() def identify_risks(self, system_context): risks = [] # Risk 1: Memory exhaustion risks.append({ 'id': 'RISK_001', 'name': 'Memory Exhaustion', 'probability': 'medium', 'impact': 'high', 'mitigation': 'Implement memory limits and monitoring', 'recovery': 'Trigger emergency stop and restart' }) # Risk 2: Infinite recursion risks.append({ 'id': 'RISK_002', 'name': 'Infinite Recursion', 'probability': 'medium', 'impact': 'medium', 'mitigation': 'Stack depth limits and iteration counters', 'recovery': 'Catch recursion limit exception and unwind' }) # Risk 3: Unauthorized data access risks.append({ 'id': 'RISK_003', 'name': 'Unauthorized Data Access', 'probability': 'low', 'impact': 'critical', 'mitigation': 'Role-based access control (RBAC)', 'recovery': 'Audit logs and data restoration' }) self.risk_register.add_risks(risks) return risks def setup_emergency_procedures(self): procedures = { 'stop_execution': self._stop_execution, 'rollback_changes': self._rollback_changes, 'notify_administrators': self._notify_administrators, 'log_incident': self._log_incident, 'restore_backups': self._restore_backups } return procedures def detect_anomaly(self, execution_metrics): anomalies = [] if execution_metrics['memory_usage'] > 0.9 * execution_metrics['memory_limit']: anomalies.append('Memory usage critical') if execution_metrics['execution_time'] > 0.8 * execution_metrics['timeout']: anomalies.append('Execution time approaching limit') if execution_metrics['cpu_usage'] > 0.95 * execution_metrics['cpu_limit']: anomalies.append('CPU usage critical') return anomalies def _stop_execution(self): # Implementation to safely stop execution pass def _rollback_changes(self): # Implementation to rollback unauthorized changes pass def _notify_administrators(self, incident): # Implementation to notify admins pass def _log_incident(self, incident): # Implementation to log incident details self.incident_log.record(incident) def _restore_backups(self): # Implementation to restore from backups pass

class AutonomousAgentBuilder: def __init__(self, agent_name, capabilities): self.name = agent_name self.capabilities = capabilities self.safety_policies = {} self.audit_log = AuditLog() def configure_agent(self): self.safety_policies = { 'max_concurrent_tasks': 5, 'max_tokens_per_task': 50000, 'memory_per_task_mb': 512, 'max_api_calls_per_minute': 30, 'allowed_file_operations': ['read', 'list'], 'forbidden_operations': ['delete', 'chmod', 'system_call'] } def run_end_to_end_test(self, test_case): test_result = { 'test_id': test_case['id'], 'start_time': time.time(), 'steps': [], 'passed': False, 'errors': [] } try: # Initialize test environment test_env = self._setup_test_environment(test_case) # Execute agent result = self._execute_agent(test_env, test_case) test_result['steps'].append(f'Agent execution completed') # Validate results validation = self._validate_results(result, test_case) test_result['steps'].append(f'Validation: {validation["status"]}') # Check for safety violations violations = self._check_safety_violations(result) if violations: test_result['errors'].extend(violations) else: test_result['passed'] = True # Cleanup self._cleanup_test_environment(test_env) except Exception as e: test_result['errors'].append(str(e)) test_result['end_time'] = time.time() test_result['duration_seconds'] = test_result['end_time'] - test_result['start_time'] self.audit_log.record_test(test_result) return test_result def analyze_failure(self, test_result): analysis = { 'test_id': test_result['test_id'], 'failure_type': self._determine_failure_type(test_result), 'root_cause': None, 'contributing_factors': [], 'recommendations': [] } for error in test_result['errors']: if 'timeout' in error.lower(): analysis['failure_type'] = 'timeout' analysis['recommendations'].append('Increase timeout or optimize agent logic') elif 'memory' in error.lower(): analysis['failure_type'] = 'memory_error' analysis['recommendations'].append('Reduce memory usage or increase limit') elif 'permission' in error.lower(): analysis['failure_type'] = 'permission_denied' analysis['recommendations'].append('Review RBAC policies') return analysis def post_mortem_review(self, incident): review = { 'incident_id': incident['id'], 'summary': incident['description'], 'timeline': self._extract_timeline(incident), 'root_causes': self._identify_root_causes(incident), 'preventive_measures': self._suggest_measures(incident), 'lessons_learned': [] } review['lessons_learned'].append( 'Autonomous agents require continuous monitoring' ) review['lessons_learned'].append( 'Clear policies must be enforced from the start' ) return review def _setup_test_environment(self, test_case): # Initialize test environment with isolated resources return {'id': uuid.uuid4(), 'data': test_case['input_data']} def _execute_agent(self, env, test_case): # Execute the agent with monitoring return {'status': 'completed', 'output': test_case['expected_output']} def _validate_results(self, result, test_case): # Validate agent output against expected results return {'status': 'valid'} if result['status'] == 'completed' else {'status': 'invalid'} def _check_safety_violations(self, result): # Check for any safety violations in execution return [] def _cleanup_test_environment(self, env): # Clean up test resources pass def _determine_failure_type(self, result): return 'unknown' def _extract_timeline(self, incident): return [] def _identify_root_causes(self, incident): return [] def _suggest_measures(self, incident): return []