class MLDeploymentOrchestrator:
    def __init__(self):
        self.model_registry = ModelRegistry()
        self.feature_pipeline = FeaturePipelineManager()
        self.inference_services = InferenceServiceManager()
        self.traffic_manager = TrafficManager()
        self.validation_suite = ModelValidationSuite()
        self.rollback_manager = RollbackManager()

    async def deploy_model_update(self, deployment_config):
        """Orchestrate zero-downtime ML model deployment"""
        deployment_id = generate_deployment_id()

        try:
            # Phase 1: Pre-deployment validation
            validation_result = await self.pre_deployment_validation(deployment_config)
            if not validation_result.passed:
                raise DeploymentValidationError(validation_result.errors)

            # Phase 2: Prepare new model version
            prepared_artifacts = await self.prepare_model_artifacts(deployment_config)

            # Phase 3: Stage deployment across infrastructure
            staging_result = await self.stage_deployment(prepared_artifacts)

            # Phase 4: Gradual traffic shift with monitoring
            traffic_shift_result = await self.execute_gradual_rollout(
                deployment_config, staging_result
            )

            # Phase 5: Post-deployment validation
            post_validation = await self.post_deployment_validation(deployment_id)

            # Phase 6: Cleanup old versions
            await self.cleanup_old_versions(deployment_config)

            return DeploymentResult(
                deployment_id=deployment_id,
                status='success',
                metrics=traffic_shift_result.metrics
            )

        except Exception as e:
            # Automatic rollback on any failure
            await self.rollback_manager.execute_rollback(deployment_id, str(e))
            raise DeploymentFailedException(f"Deployment failed: {e}")

    async def pre_deployment_validation(self, config):
        """Comprehensive validation before deployment starts"""
        validation_tasks = [
            self.validate_model_artifacts(config.model_artifacts),
            self.validate_feature_compatibility(config.feature_schema),
            self.validate_infrastructure_readiness(),
            self.validate_traffic_capacity(config.expected_load),
            self.validate_rollback_readiness(config.current_version)
        ]

        results = await asyncio.gather(*validation_tasks, return_exceptions=True)

        validation_result = ValidationResult()
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                validation_result.add_error(f"Validation task {i} failed: {result}")
            elif not result.passed:
                validation_result.add_error(result.error_message)

        return validation_result

class ModelArtifactManager:
    def __init__(self):
        self.storage_backends = {
            's3': S3ModelStorage(),
            'gcs': GCSModelStorage(), 
            'azure': AzureModelStorage()
        }
        self.artifact_cache = ArtifactCache()

    async def prepare_model_artifacts(self, model_config):
        """Prepare and distribute model artifacts across regions"""
        preparation_tasks = []

        # Download and validate model artifacts
        model_artifacts = await self.download_model_artifacts(model_config.model_uri)
        validation_result = await self.validate_model_integrity(model_artifacts)

        if not validation_result.valid:
            raise ArtifactValidationError(validation_result.errors)

        # Pre-warm model artifacts in all target regions
        for region in model_config.target_regions:
            task = self.pre_warm_artifacts_in_region(model_artifacts, region)
            preparation_tasks.append(task)

        # Pre-compile model for target hardware if needed
        if model_config.requires_compilation:
            compilation_tasks = [
                self.compile_for_hardware(model_artifacts, hw_config) 
                for hw_config in model_config.hardware_targets
            ]
            preparation_tasks.extend(compilation_tasks)

        await asyncio.gather(*preparation_tasks)

        return PreparedArtifacts(
            model_artifacts=model_artifacts,
            regions=model_config.target_regions,
            compilation_results=compilation_tasks if model_config.requires_compilation else None
        )

    async def validate_model_integrity(self, artifacts):
        """Validate model artifacts integrity and compatibility"""
        validations = ValidationResult()

        # Check file integrity
        for artifact in artifacts.files:
            computed_hash = await self.compute_file_hash(artifact.path)
            if computed_hash != artifact.expected_hash:
                validations.add_error(f"Hash mismatch for {artifact.path}")

        # Validate model can be loaded
        try:
            test_model = await self.load_model_for_testing(artifacts)
            test_prediction = await test_model.predict(artifacts.sample_input)

            if not self.is_valid_prediction_format(test_prediction):
                validations.add_error("Model produces invalid prediction format")

        except Exception as e:
            validations.add_error(f"Model loading failed: {e}")

        # Check compatibility with current feature pipeline
        schema_compatibility = await self.check_feature_schema_compatibility(
            artifacts.feature_schema
        )
        if not schema_compatibility.compatible:
            validations.add_error(f"Feature schema incompatible: {schema_compatibility.issues}")

        return validations

class MLCanaryDeployment:
    def __init__(self):
        self.traffic_splitter = TrafficSplitter()
        self.metrics_collector = MLMetricsCollector()
        self.decision_engine = CanaryDecisionEngine()
        self.safety_guards = SafetyGuards()

    async def execute_canary_rollout(self, canary_config):
        """Execute canary deployment with ML-specific validation"""
        canary_session = CanarySession(
            canary_version=canary_config.new_version,
            control_version=canary_config.current_version,
            traffic_stages=canary_config.traffic_stages
        )

        for stage in canary_config.traffic_stages:
            # Configure traffic split
            await self.traffic_splitter.set_traffic_split(
                canary_percentage=stage.canary_percentage,
                control_percentage=100 - stage.canary_percentage
            )

            # Monitor stage for specified duration
            stage_result = await self.monitor_canary_stage(
                canary_session, stage
            )

            # Make go/no-go decision
            decision = await self.decision_engine.evaluate_stage(stage_result)

            if decision.action == 'abort':
                await self.abort_canary_deployment(canary_session, decision.reason)
                raise CanaryDeploymentAborted(decision.reason)
            elif decision.action == 'hold':
                # Wait for more data before proceeding
                additional_monitoring = await self.extended_monitoring(
                    canary_session, stage, decision.hold_duration
                )
                decision = await self.decision_engine.evaluate_stage(additional_monitoring)

            if decision.action != 'proceed':
                await self.abort_canary_deployment(canary_session, decision.reason)
                raise CanaryDeploymentAborted(decision.reason)

        # Canary successful, complete rollout
        await self.complete_canary_rollout(canary_session)
        return canary_session.get_final_metrics()

    async def monitor_canary_stage(self, canary_session, stage):
        """Monitor canary stage with comprehensive ML metrics"""
        monitoring_duration = stage.duration
        collection_interval = 30  # seconds

        stage_metrics = {
            'technical_metrics': defaultdict(list),
            'ml_metrics': defaultdict(list),
            'business_metrics': defaultdict(list),
            'safety_violations': []
        }

        start_time = time.time()

        while time.time() - start_time < monitoring_duration:
            # Collect current metrics
            current_metrics = await self.collect_current_metrics(canary_session)

            # Technical metrics (latency, errors, throughput)
            for metric, value in current_metrics.technical.items():
                stage_metrics['technical_metrics'][metric].append(value)

            # ML-specific metrics (prediction quality, confidence, drift)
            for metric, value in current_metrics.ml.items():
                stage_metrics['ml_metrics'][metric].append(value)

            # Business metrics (conversion, revenue, user engagement)
            for metric, value in current_metrics.business.items():
                stage_metrics['business_metrics'][metric].append(value)

            # Check safety guards
            safety_check = await self.safety_guards.check_safety_conditions(current_metrics)
            if not safety_check.safe:
                stage_metrics['safety_violations'].append(safety_check.violations)
                # Immediate abort on safety violations
                return StageResult(status='abort', reason='safety_violation', metrics=stage_metrics)

            await asyncio.sleep(collection_interval)

        return StageResult(status='completed', metrics=stage_metrics)

class MLMetricsCollector:
    def __init__(self):
        self.prediction_tracker = PredictionTracker()
        self.performance_analyzer = ModelPerformanceAnalyzer()
        self.drift_detector = DriftDetector()

    async def collect_current_metrics(self, canary_session):
        """Collect comprehensive metrics for both canary and control"""
        # Get recent predictions from both versions
        canary_predictions = await self.prediction_tracker.get_recent_predictions(
            version=canary_session.canary_version,
            time_window=300  # Last 5 minutes
        )

        control_predictions = await self.prediction_tracker.get_recent_predictions(
            version=canary_session.control_version,
            time_window=300
        )

        metrics = MLMetrics()

        # Technical metrics
        metrics.technical = {
            'canary_latency_p95': np.percentile([p.latency for p in canary_predictions], 95),
            'control_latency_p95': np.percentile([p.latency for p in control_predictions], 95),
            'canary_error_rate': sum(1 for p in canary_predictions if p.error) / len(canary_predictions),
            'control_error_rate': sum(1 for p in control_predictions if p.error) / len(control_predictions),
            'canary_throughput': len(canary_predictions) / 300,  # QPS
            'control_throughput': len(control_predictions) / 300
        }

        # ML-specific metrics
        if canary_predictions and control_predictions:
            canary_confidences = [p.confidence for p in canary_predictions if p.confidence]
            control_confidences = [p.confidence for p in control_predictions if p.confidence]

            metrics.ml = {
                'canary_mean_confidence': np.mean(canary_confidences) if canary_confidences else 0,
                'control_mean_confidence': np.mean(control_confidences) if control_confidences else 0,
                'canary_low_confidence_rate': sum(1 for c in canary_confidences if c < 0.7) / len(canary_confidences) if canary_confidences else 0,
                'control_low_confidence_rate': sum(1 for c in control_confidences if c < 0.7) / len(control_confidences) if control_confidences else 0
            }

            # Drift detection between versions
            canary_features = [p.features for p in canary_predictions if p.features]
            control_features = [p.features for p in control_predictions if p.features]

            if canary_features and control_features:
                drift_score = await self.drift_detector.compare_feature_distributions(
                    canary_features, control_features
                )
                metrics.ml['prediction_drift_score'] = drift_score

        # Business metrics (when outcomes are available)
        canary_outcomes = await self.get_recent_outcomes(canary_session.canary_version)
        control_outcomes = await self.get_recent_outcomes(canary_session.control_version)

        if canary_outcomes and control_outcomes:
            metrics.business = {
                'canary_conversion_rate': np.mean([o.converted for o in canary_outcomes]),
                'control_conversion_rate': np.mean([o.converted for o in control_outcomes]),
                'canary_revenue_per_prediction': np.mean([o.revenue for o in canary_outcomes]),
                'control_revenue_per_prediction': np.mean([o.revenue for o in control_outcomes])
            }

        return metrics

class CanaryDecisionEngine:
    def __init__(self):
        self.decision_criteria = {
            'technical': TechnicalCriteria(),
            'ml_quality': MLQualityCriteria(),
            'business_impact': BusinessImpactCriteria()
        }
        self.decision_weights = {
            'technical': 0.3,
            'ml_quality': 0.4,
            'business_impact': 0.3
        }

    async def evaluate_stage(self, stage_result):
        """Make data-driven decision about canary progression"""
        if stage_result.status == 'abort':
            return CanaryDecision(action='abort', reason=stage_result.reason)

        decision_scores = {}
        decision_details = {}

        # Evaluate each criteria category
        for criteria_name, criteria in self.decision_criteria.items():
            evaluation = await criteria.evaluate(stage_result.metrics)
            decision_scores[criteria_name] = evaluation.score
            decision_details[criteria_name] = evaluation.details

            # Hard failure conditions
            if evaluation.hard_failure:
                return CanaryDecision(
                    action='abort',
                    reason=f"Hard failure in {criteria_name}: {evaluation.failure_reason}",
                    details=decision_details
                )

        # Calculate weighted overall score
        overall_score = sum(
            score * self.decision_weights[criteria]
            for criteria, score in decision_scores.items()
        )

        # Decision logic
        if overall_score >= 0.8:
            action = 'proceed'
        elif overall_score >= 0.6:
            action = 'hold'
            hold_duration = 300  # 5 minutes additional monitoring
        else:
            action = 'abort'
            reason = f"Overall score {overall_score:.2f} below threshold"

        return CanaryDecision(
            action=action,
            reason=reason if action == 'abort' else None,
            hold_duration=hold_duration if action == 'hold' else None,
            score=overall_score,
            details=decision_details
        )

class MLQualityCriteria:
    def __init__(self):
        self.thresholds = {
            'confidence_degradation_max': 0.05,  # 5% max drop in confidence
            'drift_score_max': 0.3,
            'low_confidence_rate_max': 0.15  # 15% max low confidence predictions
        }

    async def evaluate(self, metrics):
        """Evaluate ML-specific quality criteria"""
        ml_metrics = metrics.get('ml_metrics', {})
        evaluation = CriteriaEvaluation()

        # Check confidence degradation
        canary_conf = np.mean(ml_metrics.get('canary_mean_confidence', [1.0]))
        control_conf = np.mean(ml_metrics.get('control_mean_confidence', [1.0]))

        if control_conf > 0:
            confidence_degradation = (control_conf - canary_conf) / control_conf

            if confidence_degradation > self.thresholds['confidence_degradation_max']:
                evaluation.add_issue(
                    'confidence_degradation',
                    f"Confidence dropped by {confidence_degradation:.1%}",
                    severity='high'
                )

        # Check drift score
        drift_score = np.mean(ml_metrics.get('prediction_drift_score', [0]))
        if drift_score > self.thresholds['drift_score_max']:
            evaluation.add_issue(
                'high_drift',
                f"Drift score {drift_score:.2f} above threshold",
                severity='medium'
            )

        # Check low confidence rate
        canary_low_conf = np.mean(ml_metrics.get('canary_low_confidence_rate', [0]))
        if canary_low_conf > self.thresholds['low_confidence_rate_max']:
            evaluation.add_issue(
                'high_low_confidence_rate',
                f"Low confidence rate {canary_low_conf:.1%}",
                severity='medium'
            )

        # Calculate overall score
        evaluation.score = 1.0 - (len(evaluation.high_severity_issues) * 0.4 + len(evaluation.medium_severity_issues) * 0.2)
        evaluation.score = max(0, evaluation.score)

        return evaluation

class FeaturePipelineCoordinator:
    def __init__(self):
        self.pipeline_registry = FeaturePipelineRegistry()
        self.schema_validator = SchemaValidator()
        self.version_manager = FeatureVersionManager()

    async def coordinate_feature_update(self, model_deployment_config):
        """Coordinate feature pipeline updates with model deployment"""
        coordination_plan = await self.create_coordination_plan(model_deployment_config)

        try:
            # Phase 1: Deploy new feature pipeline version
            await self.deploy_feature_pipeline_version(
                coordination_plan.new_pipeline_version
            )

            # Phase 2: Run dual pipeline mode (old + new)
            await self.enable_dual_pipeline_mode(coordination_plan)

            # Phase 3: Validate feature compatibility
            validation_result = await self.validate_feature_compatibility(
                coordination_plan
            )

            if not validation_result.compatible:
                raise FeatureCompatibilityError(validation_result.issues)

            # Phase 4: Switch model to new features gradually
            await self.gradual_feature_switch(coordination_plan)

            # Phase 5: Cleanup old pipeline version
            await self.cleanup_old_pipeline_version(coordination_plan)

        except Exception as e:
            await self.rollback_feature_pipeline_changes(coordination_plan)
            raise FeaturePipelineCoordinationError(f"Feature coordination failed: {e}")

    async def create_coordination_plan(self, model_config):
        """Create detailed plan for coordinating pipeline and model updates"""
        current_pipeline = await self.pipeline_registry.get_current_version(
            model_config.model_name
        )

        required_features = model_config.required_feature_schema

        # Analyze what needs to change
        schema_diff = await self.schema_validator.compare_schemas(
            current_pipeline.output_schema,
            required_features
        )

        coordination_plan = CoordinationPlan(
            model_name=model_config.model_name,
            current_pipeline_version=current_pipeline.version,
            new_pipeline_version=model_config.feature_pipeline_version,
            schema_changes=schema_diff,
            coordination_strategy=self.determine_coordination_strategy(schema_diff)
        )

        return coordination_plan

    def determine_coordination_strategy(self, schema_diff):
        """Determine best strategy based on schema changes"""
        if schema_diff.has_breaking_changes:
            return 'staged_replacement'  # Deploy new, validate, switch, cleanup old
        elif schema_diff.has_additions:
            return 'additive_deployment'  # Add new features, maintain old ones
        else:
            return 'in_place_update'  # Safe to update in place

class DualPipelineManager:
    def __init__(self):
        self.active_pipelines = {}
        self.feature_router = FeatureRouter()

    async def enable_dual_mode(self, pipeline_a_version, pipeline_b_version):
        """Run two feature pipeline versions simultaneously"""
        # Start both pipelines
        pipeline_a = await self.start_pipeline_version(pipeline_a_version)
        pipeline_b = await self.start_pipeline_version(pipeline_b_version)

        # Configure feature router to compute features using both pipelines
        await self.feature_router.configure_dual_mode(
            primary_pipeline=pipeline_a,
            secondary_pipeline=pipeline_b
        )

        self.active_pipelines[pipeline_a_version] = pipeline_a
        self.active_pipelines[pipeline_b_version] = pipeline_b

        return DualPipelineSession(
            pipeline_a=pipeline_a,
            pipeline_b=pipeline_b,
            router=self.feature_router
        )

class FeatureRouter:
    def __init__(self):
        self.routing_config = {}
        self.feature_cache = FeatureCache()

    async def get_features_for_model(self, model_version, input_data):
        """Route feature computation based on model requirements"""
        routing_rule = self.routing_config.get(model_version)

        if not routing_rule:
            raise FeatureRoutingError(f"No routing rule for model {model_version}")

        # Check cache first
        cache_key = self.generate_cache_key(model_version, input_data)
        cached_features = await self.feature_cache.get(cache_key)

        if cached_features:
            return cached_features

        # Compute features using specified pipeline
        features = await routing_rule.pipeline.compute_features(input_data)

        # Apply any required transformations
        if routing_rule.transformations:
            features = await self.apply_transformations(
                features, routing_rule.transformations
            )

        # Cache results
        await self.feature_cache.set(cache_key, features, ttl=routing_rule.cache_ttl)

        return features

    async def configure_dual_mode(self, primary_pipeline, secondary_pipeline):
        """Configure router for dual pipeline mode with comparison"""
        self.dual_mode_config = DualModeConfig(
            primary=primary_pipeline,
            secondary=secondary_pipeline,
            comparison_enabled=True,
            comparison_sample_rate=0.1  # Compare 10% of requests
        )

    async def compute_features_dual_mode(self, input_data):
        """Compute features using both pipelines and compare results"""
        # Always compute using primary pipeline
        primary_features = await self.dual_mode_config.primary.compute_features(input_data)

        # Compute using secondary pipeline for comparison
        if random.random() < self.dual_mode_config.comparison_sample_rate:
            secondary_features = await self.dual_mode_config.secondary.compute_features(input_data)

            # Compare feature outputs
            comparison_result = await self.compare_feature_outputs(
                primary_features, secondary_features
            )

            # Log comparison for analysis
            await self.log_feature_comparison(comparison_result)

        return primary_features

class ProgressiveModelReplacer:
    def __init__(self):
        self.model_pool = ModelPool()
        self.load_balancer = ModelLoadBalancer()
        self.resource_monitor = ResourceMonitor()

    async def execute_progressive_replacement(self, replacement_config):
        """Replace models progressively across instance pool"""
        replacement_session = ProgressiveReplacementSession(
            old_version=replacement_config.current_version,
            new_version=replacement_config.new_version,
            instance_pool=self.model_pool.get_all_instances()
        )

        # Determine replacement strategy based on resource constraints
        strategy = await self.determine_replacement_strategy(replacement_config)

        if strategy == 'rolling_replacement':
            return await self.rolling_replacement(replacement_session)
        elif strategy == 'blue_green_replacement':
            return await self.blue_green_replacement(replacement_session)
        elif strategy == 'shadow_replacement':
            return await self.shadow_replacement(replacement_session)
        else:
            raise UnsupportedReplacementStrategy(strategy)

    async def rolling_replacement(self, session):
        """Replace models one instance at a time"""
        total_instances = len(session.instance_pool)
        batch_size = max(1, total_instances // 10)  # 10% at a time

        for batch_start in range(0, total_instances, batch_size):
            batch_instances = session.instance_pool[batch_start:batch_start + batch_size]

            # Remove instances from load balancer
            await self.load_balancer.remove_instances(batch_instances)

            # Wait for existing requests to complete
            await self.wait_for_request_completion(batch_instances)

            # Replace models on instances
            replacement_tasks = [
                self.replace_model_on_instance(instance, session.new_version)
                for instance in batch_instances
            ]

            await asyncio.gather(*replacement_tasks)

            # Validate new models
            validation_tasks = [
                self.validate_instance_health(instance)
                for instance in batch_instances
            ]

            validation_results = await asyncio.gather(*validation_tasks)

            # Only re-add instances that pass validation
            healthy_instances = [
                instance for instance, result in zip(batch_instances, validation_results)
                if result.healthy
            ]

            await self.load_balancer.add_instances(healthy_instances)

            # Monitor batch performance before proceeding
            batch_monitoring = await self.monitor_batch_performance(
                healthy_instances, duration=60
            )

            if not batch_monitoring.performance_acceptable:
                # Rollback this batch and abort
                await self.rollback_batch(batch_instances, session.old_version)
                raise ModelReplacementFailed("Performance degradation detected")

        return ReplacementResult(status='success', replaced_instances=total_instances)

    async def blue_green_replacement(self, session):
        """Replace using blue-green strategy with full instance duplication"""
        # Check resource availability for full duplication
        resource_check = await self.resource_monitor.check_capacity_for_duplication()
        if not resource_check.sufficient:
            raise InsufficientResourcesError("Cannot allocate resources for blue-green deployment")

        # Allocate new instances (green)
        green_instances = await self.allocate_green_instances(
            count=len(session.instance_pool),
            model_version=session.new_version
        )

        try:
            # Load new model on green instances
            loading_tasks = [
                self.load_model_on_instance(instance, session.new_version)
                for instance in green_instances
            ]

            await asyncio.gather(*loading_tasks)

            # Validate green instances
            validation_results = await self.validate_green_instances(green_instances)
            if not validation_results.all_healthy:
                raise GreenInstanceValidationFailed(validation_results.issues)

            # Gradually shift traffic from blue to green
            traffic_shift_result = await self.execute_traffic_shift(
                blue_instances=session.instance_pool,
                green_instances=green_instances
            )

            if not traffic_shift_result.successful:
                raise TrafficShiftFailed(traffic_shift_result.reason)

            # Cleanup blue instances
            await self.cleanup_blue_instances(session.instance_pool)

            return ReplacementResult(
                status='success',
                new_instances=green_instances,
                metrics=traffic_shift_result.metrics
            )

        except Exception as e:
            # Cleanup green instances on failure
            await self.cleanup_green_instances(green_instances)
            raise e

class ModelLoadBalancer:
    def __init__(self):
        self.active_instances = set()
        self.traffic_weights = {}
        self.health_checker = InstanceHealthChecker()

    async def execute_traffic_shift(self, blue_instances, green_instances):
        """Gradually shift traffic from blue to green instances"""
        shift_stages = [
            {'green_percentage': 10, 'duration': 300},   # 10% for 5 minutes
            {'green_percentage': 25, 'duration': 300},   # 25% for 5 minutes
            {'green_percentage': 50, 'duration': 600},   # 50% for 10 minutes
            {'green_percentage': 75, 'duration': 300},   # 75% for 5 minutes
            {'green_percentage': 100, 'duration': 300}   # 100% for 5 minutes
        ]

        shift_metrics = TrafficShiftMetrics()

        for stage in shift_stages:
            # Configure traffic weights
            await self.set_traffic_weights(
                blue_instances=blue_instances,
                green_instances=green_instances,
                green_percentage=stage['green_percentage']
            )

            # Monitor performance during this stage
            stage_monitoring = await self.monitor_traffic_stage(
                stage, blue_instances, green_instances
            )

            shift_metrics.add_stage_metrics(stage, stage_monitoring)

            # Check if we should abort the shift
            if stage_monitoring.performance_degraded:
                await self.abort_traffic_shift(blue_instances, green_instances)
                return TrafficShiftResult(
                    successful=False,
                    reason="Performance degradation detected",
                    aborted_at_stage=stage
                )

            # Wait for stage duration
            await asyncio.sleep(stage['duration'])

        return TrafficShiftResult(
            successful=True,
            metrics=shift_metrics
        )

    async def set_traffic_weights(self, blue_instances, green_instances, green_percentage):
        """Configure traffic distribution between instance pools"""
        blue_weight = (100 - green_percentage) / len(blue_instances)
        green_weight = green_percentage / len(green_instances)

        # Update weights for blue instances
        for instance in blue_instances:
            self.traffic_weights[instance.id] = blue_weight

        # Update weights for green instances
        for instance in green_instances:
            self.traffic_weights[instance.id] = green_weight

        # Apply configuration to load balancer
        await self.apply_traffic_configuration()

class ModelInstanceManager:
    def __init__(self):
        self.instance_registry = InstanceRegistry()
        self.resource_allocator = ResourceAllocator()
        self.model_loader = ModelLoader()

    async def replace_model_on_instance(self, instance, new_model_version):
        """Replace model on a specific instance with minimal disruption"""
        replacement_session = InstanceReplacementSession(
            instance=instance,
            old_version=instance.current_model_version,
            new_version=new_model_version
        )

        try:
            # Pre-load new model in background
            new_model = await self.model_loader.preload_model(
                instance, new_model_version
            )

            # Validate new model works correctly
            validation_result = await self.validate_model_instance(
                instance, new_model
            )

            if not validation_result.valid:
                raise ModelValidationError(validation_result.errors)

            # Atomic switch: replace old model with new one
            await self.atomic_model_switch(instance, new_model)

            # Cleanup old model resources
            await self.cleanup_old_model(instance, replacement_session.old_version)

            return InstanceReplacementResult(
                instance=instance,
                status='success',
                old_version=replacement_session.old_version,
                new_version=new_model_version
            )

        except Exception as e:
            # Cleanup any partially loaded resources
            await self.cleanup_failed_replacement(instance, new_model_version)
            raise InstanceReplacementError(f"Failed to replace model on {instance.id}: {e}")

    async def atomic_model_switch(self, instance, new_model):
        """Atomically switch from old model to new model"""
        # Use memory mapping or pointer swapping for atomic switch
        old_model_ref = instance.current_model

        # Update instance model reference atomically
        instance.current_model = new_model
        instance.current_model_version = new_model.version
        instance.last_model_update = datetime.now()

        # Wait brief period to ensure no in-flight predictions use old model
        await asyncio.sleep(0.1)

        # Schedule old model cleanup
        self.schedule_delayed_cleanup(old_model_ref, delay_seconds=30)

class ModelValidationSuite:
    def __init__(self):
        self.test_cases = TestCaseManager()
        self.performance_benchmarks = PerformanceBenchmarks()
        self.integration_tests = IntegrationTests()

    async def validate_model_instance(self, instance, model):
        """Comprehensive validation of model instance"""
        validation_result = ValidationResult()

        # Functional tests
        functional_result = await self.run_functional_tests(instance, model)
        validation_result.add_test_result('functional', functional_result)

        # Performance benchmarks
        performance_result = await self.run_performance_benchmarks(instance, model)
        validation_result.add_test_result('performance', performance_result)

        # Integration tests
        integration_result = await self.run_integration_tests(instance, model)
        validation_result.add_test_result('integration', integration_result)

        # Resource utilization check
        resource_result = await self.check_resource_utilization(instance, model)
        validation_result.add_test_result('resources', resource_result)

        return validation_result

    async def run_functional_tests(self, instance, model):
        """Run functional correctness tests"""
        test_result = TestResult('functional')

        # Load test cases
        test_cases = await self.test_cases.get_test_cases_for_model(model.name)

        for test_case in test_cases:
            try:
                # Execute prediction
                prediction = await model.predict(test_case.input_data)

                # Validate prediction format
                if not self.is_valid_prediction_format(prediction):
                    test_result.add_failure(f"Invalid prediction format for test case {test_case.id}")
                    continue

                # Check prediction correctness (if expected output provided)
                if test_case.expected_output:
                    if not self.predictions_match(prediction, test_case.expected_output, tolerance=0.1):
                        test_result.add_failure(f"Prediction mismatch for test case {test_case.id}")
                        continue

                test_result.add_success(test_case.id)

            except Exception as e:
                test_result.add_error(f"Test case {test_case.id} failed: {e}")

        return test_result

    async def run_performance_benchmarks(self, instance, model):
        """Run performance benchmarks"""
        benchmark_result = TestResult('performance')

        # Latency benchmark
        latency_results = await self.benchmark_latency(model, num_requests=100)

        if latency_results.p95_latency > self.performance_benchmarks.max_p95_latency:
            benchmark_result.add_failure(
                f"P95 latency {latency_results.p95_latency}ms exceeds threshold"
            )

        # Throughput benchmark
        throughput_results = await self.benchmark_throughput(model, duration_seconds=60)

        if throughput_results.qps < self.performance_benchmarks.min_throughput:
            benchmark_result.add_failure(
                f"Throughput {throughput_results.qps} QPS below threshold"
            )

        # Memory usage benchmark
        memory_usage = await self.measure_memory_usage(model)

        if memory_usage.peak_mb > self.performance_benchmarks.max_memory_mb:
            benchmark_result.add_failure(
                f"Memory usage {memory_usage.peak_mb}MB exceeds threshold"
            )

        return benchmark_result

class MLRollbackManager:
    def __init__(self):
        self.deployment_history = DeploymentHistory()
        self.model_registry = ModelRegistry()
        self.feature_pipeline_manager = FeaturePipelineManager()
        self.state_manager = ModelStateManager()

    async def execute_rollback(self, deployment_id, rollback_reason):
        """Execute comprehensive rollback of ML deployment"""
        rollback_session = RollbackSession(
            deployment_id=deployment_id,
            reason=rollback_reason,
            started_at=datetime.now()
        )

        try:
            # Get deployment details
            deployment = await self.deployment_history.get_deployment(deployment_id)
            if not deployment:
                raise RollbackError(f"Deployment {deployment_id} not found")

            # Determine rollback strategy
            rollback_plan = await self.create_rollback_plan(deployment)

            # Execute rollback phases
            await self.execute_rollback_phases(rollback_plan, rollback_session)

            # Validate rollback success
            validation_result = await self.validate_rollback(rollback_plan)

            if not validation_result.successful:
                raise RollbackValidationError(validation_result.issues)

            rollback_session.status = 'completed'
            rollback_session.completed_at = datetime.now()

            return rollback_session

        except Exception as e:
            rollback_session.status = 'failed'
            rollback_session.error = str(e)
            rollback_session.completed_at = datetime.now()

            # Log rollback failure for investigation
            await self.log_rollback_failure(rollback_session, e)
            raise RollbackError(f"Rollback failed: {e}")

    async def create_rollback_plan(self, deployment):
        """Create detailed rollback execution plan"""
        rollback_plan = RollbackPlan(
            deployment_id=deployment.id,
            target_version=deployment.previous_version,
            current_version=deployment.current_version
        )

        # Analyze what needs to be rolled back
        if deployment.included_model_update:
            rollback_plan.add_phase(ModelRollbackPhase(
                from_version=deployment.model_version,
                to_version=deployment.previous_model_version
            ))

        if deployment.included_feature_pipeline_update:
            rollback_plan.add_phase(FeaturePipelineRollbackPhase(
                from_version=deployment.feature_pipeline_version,
                to_version=deployment.previous_feature_pipeline_version
            ))

        if deployment.included_infrastructure_changes:
            rollback_plan.add_phase(InfrastructureRollbackPhase(
                changes=deployment.infrastructure_changes
            ))

        # Determine rollback strategy based on complexity
        rollback_plan.strategy = self.determine_rollback_strategy(rollback_plan)

        return rollback_plan

    def determine_rollback_strategy(self, rollback_plan):
        """Determine optimal rollback strategy"""
        if len(rollback_plan.phases) == 1:
            return 'simple_rollback'
        elif rollback_plan.has_breaking_changes:
            return 'staged_rollback'
        else:
            return 'coordinated_rollback'

    async def execute_rollback_phases(self, rollback_plan, rollback_session):
        """Execute rollback phases based on strategy"""
        if rollback_plan.strategy == 'simple_rollback':
            await self.execute_simple_rollback(rollback_plan, rollback_session)
        elif rollback_plan.strategy == 'staged_rollback':
            await self.execute_staged_rollback(rollback_plan, rollback_session)
        else:
            await self.execute_coordinated_rollback(rollback_plan, rollback_session)

    async def execute_coordinated_rollback(self, rollback_plan, rollback_session):
        """Execute coordinated rollback of multiple components"""
        # Phase 1: Prepare rollback artifacts
        await self.prepare_rollback_artifacts(rollback_plan)

        # Phase 2: Execute rollbacks in dependency order
        for phase in rollback_plan.get_ordered_phases():
            phase_result = await self.execute_rollback_phase(phase, rollback_session)

            if not phase_result.successful:
                # Partial rollback failure - attempt to stabilize
                await self.attempt_partial_rollback_recovery(rollback_plan, phase)
                raise RollbackPhaseError(f"Phase {phase.name} failed: {phase_result.error}")

            rollback_session.add_completed_phase(phase)

        # Phase 3: Validate system consistency
        consistency_check = await self.validate_system_consistency(rollback_plan)
        if not consistency_check.consistent:
            raise SystemInconsistencyError(consistency_check.issues)

class ModelStateManager:
    def __init__(self):
        self.state_snapshots = StateSnapshotManager()
        self.checkpoint_manager = CheckpointManager()

    async def create_deployment_checkpoint(self, deployment_config):
        """Create checkpoint before deployment for rollback"""
        checkpoint = DeploymentCheckpoint(
            deployment_id=deployment_config.deployment_id,
            timestamp=datetime.now(),
            model_state=await self.capture_model_state(deployment_config.model_name),
            feature_pipeline_state=await self.capture_pipeline_state(deployment_config.model_name),
            traffic_configuration=await self.capture_traffic_config(deployment_config.model_name)
        )

        await self.checkpoint_manager.save_checkpoint(checkpoint)
        return checkpoint

    async def restore_from_checkpoint(self, checkpoint_id):
        """Restore system state from checkpoint"""
        checkpoint = await self.checkpoint_manager.get_checkpoint(checkpoint_id)
        if not checkpoint:
            raise CheckpointNotFoundError(f"Checkpoint {checkpoint_id} not found")

        # Restore model state
        await self.restore_model_state(checkpoint.model_state)

        # Restore feature pipeline state
        await self.restore_pipeline_state(checkpoint.feature_pipeline_state)

        # Restore traffic configuration
        await self.restore_traffic_config(checkpoint.traffic_configuration)

        return RestoreResult(
            checkpoint_id=checkpoint_id,
            restored_at=datetime.now(),
            components_restored=['model', 'pipeline', 'traffic']
        )

# Automated rollback triggers
class AutomatedRollbackSystem:
    def __init__(self):
        self.rollback_manager = MLRollbackManager()
        self.monitoring_system = MLMonitoringSystem()
        self.rollback_triggers = {
            'performance_degradation': PerformanceDegradationTrigger(),
            'error_rate_spike': ErrorRateSpikeTrigger(),
            'business_metric_decline': BusinessMetricDeclineTrigger(),
            'safety_violation': SafetyViolationTrigger()
        }

    async def monitor_deployment_health(self, deployment_id):
        """Continuously monitor deployment and trigger rollback if needed"""
        monitoring_session = MonitoringSession(deployment_id)

        while monitoring_session.active:
            # Collect current metrics
            current_metrics = await self.monitoring_system.collect_metrics(deployment_id)

            # Evaluate rollback triggers
            for trigger_name, trigger in self.rollback_triggers.items():
                should_rollback = await trigger.evaluate(current_metrics)

                if should_rollback.triggered:
                    await self.execute_automated_rollback(
                        deployment_id, 
                        trigger_name, 
                        should_rollback.reason
                    )
                    monitoring_session.stop(f"Rollback triggered by {trigger_name}")
                    return

            # Wait before next check
            await asyncio.sleep(30)  # Check every 30 seconds

    async def execute_automated_rollback(self, deployment_id, trigger_name, reason):
        """Execute automated rollback with proper notifications"""
        rollback_reason = f"Automated rollback triggered by {trigger_name}: {reason}"

        # Send immediate notification
        await self.send_rollback_notification(deployment_id, rollback_reason, urgency='high')

        try:
            # Execute rollback
            rollback_result = await self.rollback_manager.execute_rollback(
                deployment_id, rollback_reason
            )

            # Send success notification
            await self.send_rollback_success_notification(deployment_id, rollback_result)

        except Exception as e:
            # Send failure notification
            await self.send_rollback_failure_notification(deployment_id, str(e))
            raise AutomatedRollbackError(f"Automated rollback failed: {e}")

class PerformanceDegradationTrigger:
    def __init__(self):
        self.thresholds = {
            'latency_p95_degradation': 0.5,  # 50% increase
            'throughput_degradation': 0.3,   # 30% decrease
            'error_rate_increase': 0.05      # 5 percentage point increase
        }

    async def evaluate(self, metrics):
        """Evaluate if performance degradation warrants rollback"""
        current_latency = metrics.get('latency_p95', 0)
        baseline_latency = metrics.get('baseline_latency_p95', current_latency)

        if baseline_latency > 0:
            latency_degradation = (current_latency - baseline_latency) / baseline_latency

            if latency_degradation > self.thresholds['latency_p95_degradation']:
                return TriggerResult(
                    triggered=True,
                    reason=f"P95 latency increased by {latency_degradation:.1%}"
                )

        # Check throughput
        current_throughput = metrics.get('throughput', 0)
        baseline_throughput = metrics.get('baseline_throughput', current_throughput)

        if baseline_throughput > 0:
            throughput_degradation = (baseline_throughput - current_throughput) / baseline_throughput

            if throughput_degradation > self.thresholds['throughput_degradation']:
                return TriggerResult(
                    triggered=True,
                    reason=f"Throughput decreased by {throughput_degradation:.1%}"
                )

        # Check error rate
        current_error_rate = metrics.get('error_rate', 0)
        baseline_error_rate = metrics.get('baseline_error_rate', 0)

        error_rate_increase = current_error_rate - baseline_error_rate

        if error_rate_increase > self.thresholds['error_rate_increase']:
            return TriggerResult(
                triggered=True,
                reason=f"Error rate increased by {error_rate_increase:.1%}"
            )

        return TriggerResult(triggered=False)