MLOps on Cloud — CI/CD for ML

A car manufacturer runs two quality checks. The design check happens when engineers submit a blueprint change — does the new design meet safety standards on paper? The production check happens on the assembly line — does the manufactured car meet standards in reality? ML CI/CD is the same: code-change tests (does the training script work correctly on a small sample?) and data-change tests (does the model trained on new production data beat the current production model?). Neither check alone is sufficient. Together they ensure nothing bad reaches customers.

The key shift in mindset: in ML, the model is not just code — it is code plus data plus hyperparameters. CI/CD must test all three dimensions simultaneously. A code change that looks fine in unit tests might produce a degraded model when combined with production data. The full pipeline test is the only reliable check.

# .github/workflows/ml-ci.yml
# Runs on every PR that touches ML code

name: ML Code CI

on:
  pull_request:
    paths:
      - 'src/**'
      - 'data/**'
      - 'requirements*.txt'
      - 'params.yaml'

env:
  PYTHON_VERSION: '3.11'

jobs:
  # ── Job 1: Unit tests ─────────────────────────────────────────────
  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: pip

      - name: Install dependencies
        run: pip install -r requirements-dev.txt

      - name: Run unit tests
        run: pytest src/ -v --tb=short --junit-xml=test-results.xml

      - name: Upload test results
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: unit-test-results
          path: test-results.xml

  # ── Job 2: Training smoke test (no cloud — small data) ────────────
  training-smoke-test:
    name: Training Smoke Test
    runs-on: ubuntu-latest
    needs: unit-tests

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: pip

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Generate small test dataset
        run: python tests/generate_test_data.py --n-rows 500 --output /tmp/test_data.csv

      - name: Run training on sample data
        run: |
          python src/train.py \
            --data-path /tmp/test_data.csv \
            --output-dir /tmp/model-output \
            --n-estimators 10 \
            --max-depth 3

      - name: Verify model output exists
        run: |
          python -c "
          import pickle, os, json
          assert os.path.exists('/tmp/model-output/model.pkl'), 'model.pkl missing'
          with open('/tmp/model-output/model.pkl', 'rb') as f:
              art = pickle.load(f)
          assert 'model' in art, 'model key missing from artifact'
          assert 'feature_cols' in art, 'feature_cols key missing'
          print('Model artifact validation passed')
          print(f'Features: {art[\"feature_cols\"]}')
          "

      - name: Verify prediction schema
        run: |
          python -c "
          import pickle, numpy as np, json
          with open('/tmp/model-output/model.pkl', 'rb') as f:
              art = pickle.load(f)
          model        = art['model']
          feature_cols = art['feature_cols']
          test_input   = np.zeros((3, len(feature_cols)))
          preds        = model.predict(test_input)
          assert preds.shape == (3,), f'Wrong output shape: {preds.shape}'
          assert all(5 <= p <= 120 for p in preds), f'Predictions out of range: {preds}'
          print(f'Schema test passed: {preds}')
          "

      - name: Check metrics are reasonable
        run: |
          python -c "
          import json
          with open('/tmp/model-output/metrics.json') as f:
              m = json.load(f)
          # On tiny 500-row sample, MAE should be computable
          assert 'train_mae' in m, 'train_mae missing from metrics'
          assert m['train_mae'] > 0, 'train_mae must be positive'
          assert m['train_mae'] < 50, f'train_mae suspiciously high: {m["train_mae"]}'
          print(f'Metrics check passed: {m}')
          "

  # ── Job 3: Code quality ───────────────────────────────────────────
  code-quality:
    name: Code Quality
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}' }

      - run: pip install ruff mypy
      - run: ruff check src/
      - run: mypy src/ --ignore-missing-imports

# .github/workflows/ml-retrain.yml
# Runs on schedule AND when triggered manually or by drift alert

name: ML Retraining Pipeline

on:
  schedule:
    - cron: '30 20 * * 0'   # Every Sunday 20:30 UTC = 4:30 PM ET
  workflow_dispatch:          # Manual trigger from GitHub UI
    inputs:
      reason:
        description: 'Reason for manual retraining'
        required: true
        default: 'drift-detected'
      force_promote:
        description: 'Force promote even if challenger is not better'
        type: boolean
        default: false
  repository_dispatch:        # Triggered by external system (drift monitor)
    types: [drift-detected, scheduled-retrain]

env:
  PYTHON_VERSION: '3.11'
  MODEL_NAME:     freshmart-delivery-time
  REGISTRY_NAME:  FreshCartDeliveryTime

jobs:
  # ── Stage 1: Unit tests (fast, no cloud) ─────────────────────────
  unit-tests:
    name: Unit Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }
      - run: pip install -r requirements-dev.txt
      - run: pytest src/ -v --tb=short

  # ── Stage 2: Cloud training job ───────────────────────────────────
  cloud-training:
    name: Cloud Training Job
    runs-on: ubuntu-latest
    needs: unit-tests
    outputs:
      job_name:      ${{ steps.submit.outputs.job_name }}
      challenger_mae: ${{ steps.metrics.outputs.mae }}
      champion_mae:   ${{ steps.compare.outputs.champion_mae }}
      should_promote: ${{ steps.compare.outputs.should_promote }}

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }

      - name: Install cloud SDK
        run: pip install azure-ai-ml azure-identity

      # Azure credentials injected from GitHub Secrets
      - name: Azure login
        uses: azure/login@v2
        with:
          creds: ${{ secrets.AZURE_BrexENTIALS }}

      - name: Submit training job
        id: submit
        run: |
          python ci/submit_training_job.py \
            --subscription-id ${{ secrets.AZURE_SUBSCRIPTION_ID }} \
            --resource-group  rg-freshmart-ml \
            --workspace       ws-freshmart-ml \
            --cutoff-date     $(date +%Y-%m-%d) \
            --output-job-name /tmp/job_name.txt

          JOB_NAME=$(cat /tmp/job_name.txt)
          echo "job_name=${JOB_NAME}" >> $GITHUB_OUTPUT
          echo "Submitted: ${JOB_NAME}"

      - name: Wait for training job
        run: |
          python ci/wait_for_job.py \
            --job-name ${{ steps.submit.outputs.job_name }} \
            --timeout  3600   # 1 hour max

      - name: Retrieve job metrics
        id: metrics
        run: |
          python ci/get_job_metrics.py \
            --job-name ${{ steps.submit.outputs.job_name }} \
            --output   /tmp/challenger_metrics.json

          MAE=$(python -c "import json; print(json.load(open('/tmp/challenger_metrics.json'))['val_mae'])")
          echo "mae=${MAE}" >> $GITHUB_OUTPUT
          echo "Challenger MAE: ${MAE}"

      - name: Compare challenger to champion
        id: compare
        run: |
          python ci/compare_models.py \
            --challenger-metrics /tmp/challenger_metrics.json \
            --registry-name      ${{ env.REGISTRY_NAME }} \
            --tolerance          0.05 \
            --output             /tmp/comparison.json

          CHAMP_MAE=$(python -c "import json; print(json.load(open('/tmp/comparison.json'))['champion_mae'])")
          PROMOTE=$(python -c "import json; print(json.load(open('/tmp/comparison.json'))['should_promote'])")
          echo "champion_mae=${CHAMP_MAE}" >> $GITHUB_OUTPUT
          echo "should_promote=${PROMOTE}" >> $GITHUB_OUTPUT

      - name: Register challenger model
        run: |
          python ci/register_model.py \
            --job-name    ${{ steps.submit.outputs.job_name }} \
            --model-name  ${{ env.MODEL_NAME }} \
            --stage       Staging

      - name: Post metrics to PR / Slack
        run: |
          python ci/notify.py \
            --challenger-mae ${{ steps.metrics.outputs.mae }} \
            --champion-mae   ${{ steps.compare.outputs.champion_mae }} \
            --should-promote ${{ steps.compare.outputs.should_promote }} \
            --slack-webhook  ${{ secrets.SLACK_WEBHOOK_ML_ALERTS }}

  # ── Stage 3: Staging tests ────────────────────────────────────────
  staging-tests:
    name: Staging Deployment Tests
    runs-on: ubuntu-latest
    needs: cloud-training
    if: needs.cloud-training.outputs.should_promote == 'True'

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }
      - run: pip install requests pytest

      - name: Deploy to staging endpoint
        run: |
          python ci/deploy_to_staging.py \
            --model-name ${{ env.MODEL_NAME }} \
            --endpoint   freshmart-delivery-staging

      - name: Run integration tests against staging
        run: |
          pytest tests/integration/ \
            --endpoint freshmart-delivery-staging \
            --n-samples 100 \
            -v

      - name: Load test staging endpoint
        run: |
          python ci/load_test.py \
            --endpoint      freshmart-delivery-staging \
            --rps            50 \
            --duration-s    30 \
            --max-p99-ms    500 \
            --max-error-pct 1.0

  # ── Stage 4: Production promotion (with optional approval gate) ───
  production-promote:
    name: Promote to Production
    runs-on: ubuntu-latest
    needs: [cloud-training, staging-tests]
    environment:
      name: production          # GitHub environment — can require manual approval
      url:  https://freshmart-delivery-endpoint.azurewebsites.net/health

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '${{ env.PYTHON_VERSION }}', cache: pip }

      - name: Gradual traffic promotion
        run: |
          python ci/gradual_promote.py \
            --model-name    ${{ env.MODEL_NAME }} \
            --endpoint      freshmart-delivery-endpoint \
            --traffic-steps "10,25,50,100" \
            --step-wait-s   600 \  # 10 min at each step
            --max-p99-ms    500 \
            --max-error-pct 1.0

      - name: Archive old production model
        run: |
          python ci/archive_old_model.py \
            --model-name ${{ env.MODEL_NAME }} \
            --keep-versions 3

      - name: Notify success
        run: |
          python ci/notify.py \
            --event promoted \
            --model-name    ${{ env.MODEL_NAME }} \
            --challenger-mae ${{ needs.cloud-training.outputs.challenger_mae }} \
            --champion-mae   ${{ needs.cloud-training.outputs.champion_mae }} \
            --slack-webhook ${{ secrets.SLACK_WEBHOOK_ML_ALERTS }}

# ci/submit_training_job.py
# Called by GitHub Actions to submit training to cloud platform

import argparse, sys, json
from datetime import datetime

def submit_azure_ml(args) -> str:
    """Submit training job to Azure ML and return job name."""
    from azure.ai.ml import MLClient, command, Input
    from azure.ai.ml.constants import AssetTypes
    from azure.identity import DefaultAzureCredential

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=args.subscription_id,
        resource_group_name=args.resource_group,
        workspace_name=args.workspace,
    )

    job = command(
        name=f'ci-retrain-{datetime.now().strftime("%Y%m%d-%H%M%S")}',
        code='./src',
        command=(
            f'python train.py '
            f'--cutoff-date {args.cutoff_date} '
            f'--output-dir ${{{{outputs.model_output}}}}'
        ),
        environment='freshmart-training-env:latest',
        compute='cpu-cluster',
        experiment_name='freshmart-delivery-ci',
        tags={'triggered_by': 'github-actions', 'cutoff': args.cutoff_date},
    )

    submitted = ml_client.jobs.create_or_update(job)
    print(f"Job submitted: {submitted.name}")
    print(f"Studio URL:    {submitted.studio_url}")

    with open(args.output_job_name, 'w') as f:
        f.write(submitted.name)

    return submitted.name

# ci/wait_for_job.py
def wait_for_azure_job(job_name: str, timeout_s: int = 3600):
    """Poll job status until complete or timeout."""
    from azure.ai.ml import MLClient
    from azure.identity import DefaultAzureCredential
    import time

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
        resource_group_name='rg-freshmart-ml',
        workspace_name='ws-freshmart-ml',
    )

    TERMINAL_STATES = {'Completed', 'Failed', 'Canceled'}
    start = time.time()

    while time.time() - start < timeout_s:
        job    = ml_client.jobs.get(job_name)
        status = job.status
        elapsed = int(time.time() - start)
        print(f"[{elapsed:>4}s] Status: {status}")

        if status in TERMINAL_STATES:
            if status != 'Completed':
                print(f"Job {status}: {job_name}")
                sys.exit(1)
            print(f"Job completed in {elapsed}s")
            return

        time.sleep(30)   # poll every 30 seconds

    print(f"Timeout after {timeout_s}s")
    sys.exit(1)

# ci/compare_models.py
def compare_models(challenger_metrics_path: str,
                    registry_name: str,
                    tolerance: float = 0.05) -> dict:
    """
    Compare challenger metrics to current production champion.
    Returns whether challenger should be promoted.
    """
    import json
    from azure.ai.ml import MLClient
    from azure.identity import DefaultAzureCredential

    with open(challenger_metrics_path) as f:
        challenger = json.load(f)
    challenger_mae = challenger['val_mae']

    ml_client = MLClient(
        credential=DefaultAzureCredential(),
        subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
        resource_group_name='rg-freshmart-ml',
        workspace_name='ws-freshmart-ml',
    )

    # Get champion metrics from model registry
    try:
        versions = ml_client.models.list(name=registry_name)
        prod_versions = [v for v in versions if v.stage == 'Production']
        if prod_versions:
            champion_run  = ml_client.jobs.get(prod_versions[0].run_id)
            champion_mae  = champion_run.outputs.get('val_mae', 999.0)
        else:
            champion_mae  = 999.0   # no champion — always promote
            print("No production champion found — challenger will be promoted")
    except Exception as e:
        champion_mae = 999.0
        print(f"Could not retrieve champion metrics: {e}")

    improvement   = (champion_mae - challenger_mae) / champion_mae * 100
    should_promote = challenger_mae <= champion_mae * (1 + tolerance)

    result = {
        'challenger_mae':  round(challenger_mae, 4),
        'champion_mae':    round(champion_mae, 4),
        'improvement_pct': round(improvement, 2),
        'should_promote':  should_promote,
        'tolerance':       tolerance,
    }

    print(f"Champion MAE:    {champion_mae:.4f}")
    print(f"Challenger MAE:  {challenger_mae:.4f}")
    print(f"Improvement:     {improvement:+.2f}%")
    print(f"Should promote:  {should_promote}")

    with open('/tmp/comparison.json', 'w') as f:
        json.dump(result, f, indent=2)

    return result

# ci/load_test.py
def load_test_endpoint(endpoint_url: str, api_key: str,
                        rps: int = 50, duration_s: int = 30,
                        max_p99_ms: float = 500.0,
                        max_error_pct: float = 1.0) -> bool:
    """
    Simple load test using concurrent requests.
    Returns True if endpoint meets SLO targets.
    """
    import threading, time, requests, statistics

    latencies = []
    errors    = []
    lock      = threading.Lock()

    sample_payload = {
        'distance_km': 3.5, 'is_peak_hour': 1,
        'order_value': 450.0, 'restaurant_avg': 33.2, 'driver_avg': 29.8,
    }

    def make_request():
        start = time.time()
        try:
            r = requests.post(
                endpoint_url,
                json=sample_payload,
                headers={'Authorization': f'Bearer {api_key}'},
                timeout=2.0,
            )
            latency = (time.time() - start) * 1000
            with lock:
                if r.status_code == 200:
                    latencies.append(latency)
                else:
                    errors.append(r.status_code)
        except Exception as e:
            with lock:
                errors.append(str(e))

    interval = 1.0 / rps
    end_time = time.time() + duration_s

    while time.time() < end_time:
        t = threading.Thread(target=make_request)
        t.daemon = True
        t.start()
        time.sleep(interval)

    time.sleep(2)   # wait for in-flight requests

    if not latencies:
        print("No successful requests — endpoint unavailable")
        return False

    p50  = statistics.median(latencies)
    p99  = sorted(latencies)[int(0.99 * len(latencies))]
    err_pct = len(errors) / (len(latencies) + len(errors)) * 100

    print(f"Load test results ({rps} RPS, {duration_s}s):")
    print(f"  Total requests:  {len(latencies) + len(errors)}")
    print(f"  p50 latency:     {p50:.1f}ms")
    print(f"  p99 latency:     {p99:.1f}ms  (max: {max_p99_ms}ms)")
    print(f"  Error rate:      {err_pct:.2f}%  (max: {max_error_pct}%)")

    passed = p99 <= max_p99_ms and err_pct <= max_error_pct
    if not passed:
        print("LOAD TEST FAILED — SLO violation")
        sys.exit(1)
    print("Load test passed")
    return True

# ci/cloud_adapter.py
# Platform-agnostic interface — one file per cloud

import os
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class JobResult:
    job_name:   str
    status:     str   # 'Completed', 'Failed'
    val_mae:    float
    val_r2:     float
    model_uri:  str   # path to model artifact

class CloudMLAdapter(ABC):
    @abstractmethod
    def submit_training_job(self, cutoff_date: str, **hyperparams) -> str:
        """Submit a training job and return the job name/ID."""

    @abstractmethod
    def wait_for_job(self, job_name: str, timeout_s: int = 3600) -> JobResult:
        """Wait for a job to complete and return results."""

    @abstractmethod
    def register_model(self, job_result: JobResult,
                        model_name: str, stage: str) -> str:
        """Register a trained model and return its version."""

    @abstractmethod
    def get_champion_metrics(self, model_name: str) -> dict:
        """Return metrics of the current production champion model."""

    @abstractmethod
    def deploy_to_endpoint(self, model_name: str, version: str,
                            endpoint: str, traffic_pct: int) -> None:
        """Deploy a model version to an endpoint with given traffic %."""

# ── Azure ML implementation ───────────────────────────────────────────
class AzureMLAdapter(CloudMLAdapter):
    def __init__(self):
        from azure.ai.ml import MLClient
        from azure.identity import DefaultAzureCredential
        self.client = MLClient(
            credential=DefaultAzureCredential(),
            subscription_id=os.environ['AZURE_SUBSCRIPTION_ID'],
            resource_group_name=os.environ['AZURE_RESOURCE_GROUP'],
            workspace_name=os.environ['AZURE_WORKSPACE'],
        )

    def submit_training_job(self, cutoff_date: str, **params) -> str:
        from azure.ai.ml import command
        job = command(
            name=f'ci-{cutoff_date}',
            code='./src',
            command=f'python train.py --cutoff-date {cutoff_date}',
            environment='freshmart-training-env:latest',
            compute='cpu-cluster',
        )
        return self.client.jobs.create_or_update(job).name

    def wait_for_job(self, job_name: str, timeout_s=3600) -> JobResult:
        import time
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            job = self.client.jobs.get(job_name)
            if job.status in ('Completed', 'Failed', 'Canceled'):
                mae = job.outputs.get('val_mae', 999.0)
                return JobResult(
                    job_name=job_name, status=job.status,
                    val_mae=float(mae), val_r2=0.0,
                    model_uri=f'azureml://jobs/{job_name}/outputs/model',
                )
            time.sleep(30)
        raise TimeoutError(f"Job {job_name} did not complete in {timeout_s}s")

    def get_champion_metrics(self, model_name: str) -> dict:
        try:
            versions = list(self.client.models.list(name=model_name))
            prod = [v for v in versions if getattr(v, 'stage', '') == 'Production']
            if not prod:
                return {'val_mae': 999.0}
            # In practice: retrieve metrics from the run that produced this model
            return {'val_mae': float(prod[0].tags.get('val_mae', 999.0))}
        except Exception:
            return {'val_mae': 999.0}

    def register_model(self, job_result, model_name, stage='Staging') -> str:
        from azure.ai.ml.entities import Model
        from azure.ai.ml.constants import AssetTypes
        model = self.client.models.create_or_update(Model(
            name=model_name,
            path=job_result.model_uri,
            type=AssetTypes.MLFLOW_MODEL,
            tags={'val_mae': str(job_result.val_mae), 'stage': stage},
        ))
        return str(model.version)

    def deploy_to_endpoint(self, model_name, version, endpoint, traffic_pct):
        print(f"AML: deploying {model_name}:{version} to {endpoint} ({traffic_pct}%)")

# ── SageMaker implementation ──────────────────────────────────────────
class SageMakerAdapter(CloudMLAdapter):
    def __init__(self):
        import boto3, sagemaker
        self.sm        = boto3.client('sagemaker')
        self.session   = sagemaker.Session()
        self.role      = sagemaker.get_execution_role()
        self.bucket    = self.session.default_bucket()

    def submit_training_job(self, cutoff_date: str, **params) -> str:
        from sagemaker.sklearn.estimator import SKLearn
        est = SKLearn(
            entry_point='train.py', source_dir='./src',
            framework_version='1.2-1', instance_type='ml.m5.large',
            role=self.role, hyperparameters={'cutoff-date': cutoff_date},
            output_path=f's3://{self.bucket}/model-output/',
            base_job_name='ci-retrain',
        )
        est.fit(inputs={'train': f's3://{self.bucket}/training-data/'}, wait=False)
        return est.latest_training_job.name

    def wait_for_job(self, job_name: str, timeout_s=3600) -> JobResult:
        import time
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            desc   = self.sm.describe_training_job(TrainingJobName=job_name)
            status = desc['TrainingJobStatus']
            if status in ('Completed', 'Failed', 'Stopped'):
                model_uri = desc.get('ModelArtifacts', {}).get('S3ModelArtifacts', '')
                mae = float(desc.get('FinalMetricDataList', [{}])[0].get('Value', 999.0))
                return JobResult(job_name, status, mae, 0.0, model_uri)
            time.sleep(30)
        raise TimeoutError(f"SageMaker job {job_name} timed out")

    def get_champion_metrics(self, model_name: str) -> dict:
        try:
            pkgs = self.sm.list_model_packages(
                ModelPackageGroupName=model_name,
                ModelApprovalStatus='Approved',
            )['ModelPackageSummaryList']
            if not pkgs:
                return {'val_mae': 999.0}
            return {'val_mae': 999.0}   # retrieve from package metadata
        except Exception:
            return {'val_mae': 999.0}

    def register_model(self, job_result, model_name, stage='Staging') -> str:
        print(f"SageMaker: registering {model_name} from {job_result.model_uri}")
        return '1'

    def deploy_to_endpoint(self, model_name, version, endpoint, traffic_pct):
        print(f"SageMaker: deploying {model_name}:{version} to {endpoint} ({traffic_pct}%)")

# ── Factory — choose adapter from environment variable ────────────────
def get_adapter() -> CloudMLAdapter:
    platform = os.environ.get('ML_PLATFORM', 'azure').lower()
    if platform == 'azure':
        return AzureMLAdapter()
    elif platform == 'sagemaker':
        return SageMakerAdapter()
    else:
        raise ValueError(f"Unknown ML_PLATFORM: {platform}")

# ── Unified CI script — platform selected by env var ─────────────────
if __name__ == '__main__':
    import sys
    adapter     = get_adapter()
    cutoff_date = sys.argv[1] if len(sys.argv) > 1 else '2026-03-28'

    job_name   = adapter.submit_training_job(cutoff_date)
    job_result = adapter.wait_for_job(job_name)

    if job_result.status != 'Completed':
        print(f"Training failed: {job_result.status}")
        sys.exit(1)

    champion   = adapter.get_champion_metrics('freshmart-delivery-time')
    promote    = job_result.val_mae <= champion['val_mae'] * 1.05

    print(f"Challenger: {job_result.val_mae:.4f}")
    print(f"Champion:   {champion['val_mae']:.4f}")
    print(f"Promote:    {promote}")

    if promote:
        version = adapter.register_model(job_result, 'freshmart-delivery-time')
        adapter.deploy_to_endpoint('freshmart-delivery-time', version,
                                    'freshmart-delivery-staging', 100)
    sys.exit(0 if promote else 2)   # exit 2 = no improvement, not a failure

SageMaker, Vertex AI, and Azure ML each cover training, a registry, and deployment — the pieces vendors demo. What they rarely cover well is the orchestration that already runs your organisation's other data pipelines, the CI system the rest of engineering uses, the incident-response tooling on-call already lives in, and the specific audit or secrets requirements security already mandated. Real stacks end up using the vendor platform for the piece it is genuinely good at — usually training compute and the registry — and gluing everything else on top with the organisation's existing tools, exactly the pattern the CloudMLAdapter interface in this module is built to support.

Buying a platform gets you compute, a registry API, and a dashboard — none of which, by themselves, decide what quality gate a challenger model must clear, how much traffic shifts to it on day one, or who gets paged when the gate fails. Those decisions are organisational practice, encoded as the specific tolerance values, promotion rules, and rollback thresholds in this module's CI workflows. Two teams running the identical SageMaker setup can have completely different MLOps maturity depending on whether those practices exist, are enforced automatically, and are actually followed under deadline pressure.

Multi-cloud MLOps trades one risk for several others: every pipeline script now needs an adapter per platform (as cloud_adapter.py shows), every engineer needs at least working familiarity with two sets of IAM, networking, and quota systems, and every incident now has two places to check instead of one. It can be the right call when a single vendor's outage history, pricing, or GPU availability genuinely threatens the business — but that is a deliberate tradeoff against real added operational complexity, not a free upgrade to reliability.

The four-stage pipeline in this module gates on metrics it was told to check — MAE within tolerance, p99 latency, error rate. It cannot catch a challenger that is numerically better but has learned a new, unintended bias, or one trained on a data window that quietly excluded a whole customer segment. That is exactly why production-promote in this module's workflow example runs against a GitHub environment that can require manual approval — automating the mechanical checks frees humans to spend their attention on the judgment calls the pipeline cannot make, it does not remove the need for that judgment.

The retraining workflow in this module triggers on a schedule or a drift alert precisely because the data feeding the model changes continuously, and a deployment-only view of MLOps would have no mechanism to notice that. In practice, the data collection and cleaning pipelines from earlier modules are just as much a part of the MLOps surface as the CI/CD workflow — a schema change upstream or a silent drop in feature freshness breaks the model just as thoroughly as a bad deployment, and neither shows up if monitoring only watches the serving endpoint.

I default to the managed service for anything that is genuinely hard infrastructure with no product-specific value in owning it — distributed training compute, GPU scheduling, managed endpoints for simple serving. I build in-house, or adapt an existing internal tool, for anything that encodes organisation-specific practice: promotion rules, quality gates, alerting thresholds, and the orchestration that ties training to the rest of the data platform. The dividing line is roughly: undifferentiated heavy lifting goes to the vendor, decisions that reflect how this specific team wants to operate stay in code the team owns.

The upside is real: one IAM model, one billing relationship, native integration between training, registry, and serving, and less adapter code to maintain. The cost is lock-in — migrating off SageMaker-specific pipeline definitions or Vertex-specific feature store APIs later is expensive, pricing negotiating leverage drops once switching is painful, and a regional outage on that vendor takes down the entire ML stack with no fallback. I would frame the decision around how much the organisation's other infrastructure is already committed to that cloud — going single-vendor on ML while everything else is multi-cloud is a different risk profile than going all-in on one vendor everywhere.

I would resist standing up a full four-stage pipeline on day one. Minimal viable looks like: training jobs submitted through whichever cloud SDK is cheapest for the team's compute needs, a single GitHub Actions workflow that runs unit tests and a smoke test on every PR, model versions tracked in something as simple as MLflow or even a naming convention in cloud storage, and one manual approval step before anything reaches production. The full automated four-stage pipeline in this module is worth building once retraining happens often enough that manual promotion is the actual bottleneck — not before.

The biggest risk is a challenger that passes every automated metric gate but is wrong in a way the gates were never told to check for — a subtle new bias, a data window that silently dropped a segment, a metric that looks better only because the evaluation set itself drifted. I mitigate this with gradual traffic promotion instead of an instant switch (10 percent, then 50, then 100, as this module's production-promote stage does), automatic rollback on any latency or error regression, and a required manual approval gate on the production environment so a human reviews the comparison numbers before the last step, even though every earlier step is fully automated.

I would look at what already runs the rest of the company's data workflows. If Airflow already orchestrates the data warehouse loads and feature pipelines, adding ML training as another set of Airflow DAGs keeps one orchestration system, one on-call rotation, and one place to see the whole data lineage — worth more than the tighter native integration a vendor's own orchestrator offers. I would lean toward the vendor's managed orchestrator only for a team with no existing orchestration investment, where the tighter coupling to that one cloud's training and registry APIs is a net simplification rather than one more system to learn.