--- name: model-registry-governance description: Establish model registry standards, governance controls, metadata schemas, approvals, and lifecycle policies for enterprise AI deployments. category: devops risk: critical source: https://github.com/BagelHole/DevOps-Security-Agent-Skills source_repo: BagelHole/DevOps-Security-Agent-Skills source_type: community date_added: '2026-09-20' license: MIT license_source: https://github.com/BagelHole/DevOps-Security-Agent-Skills/blob/main/LICENSE compatibility: Requires the relevant platform CLIs (kubectl, helm, terraform, git, CI runners) and authorized access to the target environment. Docs-only; helper scripts and templates not bundled. metadata: author: devops-skills version: '1.0' --- # Model Registry Governance Create a trustworthy system of record for model artifacts, prompts, adapters, and evaluation evidence. ## When to Use This Skill - Setting up a centralized model registry for your organization - Defining metadata standards for model artifacts - Building approval workflows for model promotion to production - Implementing lifecycle policies for model retirement - Preparing for compliance audits of AI systems ## Prerequisites - MLflow Tracking Server or Weights & Biases instance deployed - Object storage for model artifacts (S3, GCS, or MinIO) - CI/CD pipeline with access to the registry API - OPA or similar policy engine for governance checks - Git repository for policy definitions and promotion scripts ## Core Principles - **Traceability**: every production model maps to source code, data snapshot, and evaluation results. - **Reproducibility**: builds are deterministic with pinned dependencies. - **Policy-driven promotion**: no manual bypass for critical safety checks. - **Lifecycle hygiene**: stale, vulnerable, or unowned models are retired automatically. ## MLflow Registry Setup Local development illustration only: choose a supported pinned MLflow image and compatible backend drivers before use. Do not expose an unauthenticated tracking/artifact server; production needs TLS, authentication, scoped storage credentials and registry write permissions. The Compose service binds inside its container to 0.0.0.0 while only publishing loopback on the host. Supply DB credentials from a secret provider, never commit them. Registry tags and caller-supplied approver names are mutable metadata, not signed authorization. The snippets below do not call OPA or enforce the approval workflow; enforce decisions in a protected service with authenticated approvers and immutable audit records before permitting any state change. ```bash # Install MLflow with required backends pip install mlflow[extras] psycopg2-binary boto3 # Start MLflow tracking server with PostgreSQL backend and S3 artifact store mlflow server \ --backend-store-uri "${MLFLOW_BACKEND_STORE_URI:?Set a secret-backed PostgreSQL URI}" \ --default-artifact-root s3://mlflow-artifacts/models \ --host 127.0.0.1 \ --port 5000 \ --serve-artifacts ``` ```yaml # docker-compose.yaml for MLflow services: mlflow: image: ghcr.io/mlflow/mlflow:2.12.0 command: > mlflow server --backend-store-uri postgresql://mlflow:${DB_PASSWORD}@db:5432/mlflow --default-artifact-root s3://mlflow-artifacts/models --host 0.0.0.0 --port 5000 --serve-artifacts ports: - "127.0.0.1:5000:5000" environment: AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} depends_on: - db db: image: postgres:16-alpine environment: POSTGRES_DB: mlflow POSTGRES_USER: mlflow POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata: ``` ## Required Metadata Schema ```python # model_metadata_schema.py from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime from enum import Enum class LifecycleState(str, Enum): DRAFT = "draft" CANDIDATE = "candidate" APPROVED = "approved" DEPRECATED = "deprecated" RETIRED = "retired" class RiskRating(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class ModelMetadata(BaseModel): """Required metadata for every registered model.""" # Identity name: str = Field(description="Model name matching registry key") version: str = Field(description="Semantic version") checksum: str = Field(description="SHA-256 of model artifact") storage_uri: str = Field(description="Artifact store path") # Lineage base_model: str = Field(description="Parent model identifier") fine_tune_method: Optional[str] = Field(default=None) training_dataset: Optional[str] = Field(default=None) training_date: Optional[datetime] = Field(default=None) source_commit: str = Field(description="Git SHA of training code") # Evaluation eval_datasets: List[str] = Field(description="Evaluation dataset IDs") eval_report_uri: str = Field(description="Path to evaluation results") quality_score: float = Field(ge=0, le=1) safety_score: float = Field(ge=0, le=1) # Governance license: str = Field(description="SPDX license identifier") allowed_use_cases: List[str] prohibited_use_cases: List[str] risk_rating: RiskRating security_controls: List[str] # Ownership owner: str = Field(description="Primary owner email") backup_owner: str = Field(description="Backup owner email") escalation_contact: str team: str # Lifecycle state: LifecycleState = LifecycleState.DRAFT created_at: datetime = Field(default_factory=datetime.utcnow) approved_at: Optional[datetime] = None approved_by: Optional[str] = None expires_at: Optional[datetime] = None ``` ## Model Registration Script ```python # register_model.py import mlflow from mlflow.tracking import MlflowClient import json import hashlib def register_model( model_path: str, model_name: str, metadata: dict, mlflow_uri: str = "http://mlflow:5000" ): """Register a model with full metadata and governance tags.""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient() # Validate metadata first (Pydantic v2); never treat arbitrary dict fields as approved. metadata = ModelMetadata.model_validate(metadata).model_dump(mode="json") # Compute artifact checksum with open(model_path, "rb") as f: checksum = hashlib.sha256(f.read()).hexdigest() metadata["checksum"] = checksum # Log model with metadata with mlflow.start_run(run_name=f"register-{model_name}-{metadata['version']}") as run: # Log all metadata as params mlflow.log_params({ "model_name": model_name, "version": metadata["version"], "base_model": metadata["base_model"], "risk_rating": metadata["risk_rating"], "owner": metadata["owner"], "license": metadata["license"], }) # Log quality metrics mlflow.log_metrics({ "quality_score": metadata["quality_score"], "safety_score": metadata["safety_score"], }) # Log full metadata as artifact with open("metadata.json", "w") as f: json.dump(metadata, f, indent=2, default=str) mlflow.log_artifact("metadata.json") # Log model artifact mlflow.log_artifact(model_path) # Register in model registry # model_path here is a single-file artifact, not necessarily an MLflow model. # Package/log a real model with its flavor log_model API first, then use its returned URI. raise NotImplementedError("Supply the URI of a flavor-logged MLflow model before registry creation") model_uri = "" result = mlflow.register_model(model_uri, model_name) # Set lifecycle tags client.set_model_version_tag( model_name, result.version, "state", "draft" ) client.set_model_version_tag( model_name, result.version, "risk_rating", metadata["risk_rating"] ) client.set_model_version_tag( model_name, result.version, "checksum", checksum ) return result ``` ## Approval Workflow 1. Registration request created from CI. 2. Security checks (artifact scan, dependency scan, provenance). 3. Evaluation package uploaded (quality, toxicity, jailbreak, bias, latency, cost). 4. Required approvals: platform + product + security (as policy dictates). 5. Promotion to stage/prod based on signed decision record. ## Promotion Script ```python # promote_model.py import mlflow from mlflow.tracking import MlflowClient from datetime import datetime import sys def promote_model( model_name: str, version: str, target_stage: str, approver: str, mlflow_uri: str = "http://mlflow:5000" ): """Promote a model version after governance checks pass.""" mlflow.set_tracking_uri(mlflow_uri) client = MlflowClient() # Verify current state allows promotion mv = client.get_model_version(model_name, version) current_state = mv.tags.get("state", "draft") valid_transitions = { "draft": ["candidate"], "candidate": ["approved", "draft"], "approved": ["deprecated"], "deprecated": ["retired"], } if target_stage not in valid_transitions.get(current_state, []): raise ValueError( f"Invalid transition: {current_state} -> {target_stage}. " f"Allowed: {valid_transitions.get(current_state, [])}" ) # Verify required eval scores for production promotion if target_stage == "approved": run = client.get_run(mv.run_id) quality = float(run.data.metrics.get("quality_score", 0)) safety = float(run.data.metrics.get("safety_score", 0)) if quality < 0.85: raise ValueError(f"Quality score {quality} below threshold 0.85") if safety < 0.95: raise ValueError(f"Safety score {safety} below threshold 0.95") # Record promotion now = datetime.utcnow().isoformat() client.set_model_version_tag(model_name, version, "state", target_stage) client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_at", now) client.set_model_version_tag(model_name, version, f"promoted_to_{target_stage}_by", approver) # Legacy MLflow stage API (not an alias); deprecated since MLflow 2.9. # Prefer explicit aliases plus environment access controls for new integrations. # Transition legacy stage stage_map = { "candidate": "Staging", "approved": "Production", "deprecated": "Archived", } if target_stage in stage_map: client.transition_model_version_stage( model_name, version, stage_map[target_stage] ) print(f"Model {model_name} v{version}: {current_state} -> {target_stage}") print(f"Approved by: {approver} at {now}") if __name__ == "__main__": promote_model( model_name=sys.argv[1], version=sys.argv[2], target_stage=sys.argv[3], approver=sys.argv[4], ) ``` ## Lifecycle States | State | Description | Serving Allowed | New Usage | |-------|-------------|-----------------|-----------| | `draft` | Internal experimentation | Dev only | Dev only | | `candidate` | Passed baseline tests | Staging | Staging | | `approved` | Authorized for production | All environments | Yes | | `deprecated` | Replacement announced | Existing only | Blocked | | `retired` | Archived for audit | None | None | ## Lifecycle Automation ```python # lifecycle_policy.py from mlflow.tracking import MlflowClient from datetime import datetime, timedelta def enforce_lifecycle_policies(mlflow_uri: str = "http://mlflow:5000"): """Run periodic lifecycle enforcement.""" client = MlflowClient(tracking_uri=mlflow_uri) for rm in client.search_registered_models(): for mv in client.search_model_versions(f"name='{rm.name}'"): tags = mv.tags state = tags.get("state", "draft") # Auto-deprecate models with expired approvals (90 days) if state == "approved": approved_at = tags.get("promoted_to_approved_at") if approved_at: approved_date = datetime.fromisoformat(approved_at) if datetime.utcnow() - approved_date > timedelta(days=90): print(f"Auto-deprecating {rm.name} v{mv.version}: approval expired") client.set_model_version_tag(rm.name, mv.version, "state", "deprecated") client.set_model_version_tag(rm.name, mv.version, "promoted_to_deprecated_at", datetime.utcnow().isoformat()) client.set_model_version_tag( rm.name, mv.version, "auto_deprecated_reason", "approval_expired" ) # Auto-retire deprecated models after 30 days if state == "deprecated": deprecated_at = tags.get("promoted_to_deprecated_at") if deprecated_at: deprecated_date = datetime.fromisoformat(deprecated_at) if datetime.utcnow() - deprecated_date > timedelta(days=30): print(f"Auto-retiring {rm.name} v{mv.version}") client.set_model_version_tag(rm.name, mv.version, "state", "retired") client.transition_model_version_stage( rm.name, mv.version, "Archived" ) # Flag drafts with no activity for 14 days if state == "draft": created = datetime.utcfromtimestamp(mv.creation_timestamp / 1000) if datetime.utcnow() - created > timedelta(days=14): print(f"Stale draft: {rm.name} v{mv.version}") ``` The lifecycle loop only updates registry metadata: it does not stop serving, revoke aliases or prove that consumers migrated. Plan and verify those actions before retirement. Serialize promotions and evaluate fail-closed policy evidence bound to the exact artifact digest; example score thresholds are illustrative, not universal safety certification. ## Governance Policies (OPA/Rego) Rego v1 illustration. A production admission service must reject malformed/missing scan, evaluation and approval inputs before evaluating these deny rules. `sbom_uri` must also be part of the chosen metadata schema; the minimal schema above does not yet include it. ```rego # policy/model_governance.rego package model.governance import rego.v1 # Reject artifacts without SBOM deny contains msg if { not input.metadata.sbom_uri msg := "Model must include SBOM artifact URI" } # Block promotion if critical CVEs remain deny contains msg if { input.target_state == "approved" input.security_scan.critical_cves > 0 msg := sprintf("Cannot promote: %d critical CVEs unresolved", [input.security_scan.critical_cves]) } # Require refreshed evals after prompt changes deny contains msg if { input.target_state == "approved" input.prompt_changed not input.eval_refreshed_after_prompt_change msg := "Evaluation must be re-run after prompt template changes" } # Require minimum eval scores for production deny contains msg if { input.target_state == "approved" input.metadata.quality_score < 0.85 msg := sprintf("Quality score %.2f below threshold 0.85", [input.metadata.quality_score]) } # Require dual approval for high-risk models deny contains msg if { input.target_state == "approved" input.metadata.risk_rating in {"high", "critical"} count(input.approvals) < 2 msg := "High-risk models require at least 2 approvals" } ``` ## Audit Readiness Maintain immutable records of: - Who approved and when - Which policy checks executed - Which exceptions were granted - What model/version served each customer request window ## Troubleshooting | Issue | Diagnosis | Resolution | |-------|-----------|------------| | Model registration fails | Check MLflow server connectivity and artifact store permissions | Verify S3/GCS credentials and bucket policy | | Promotion blocked by policy | Review OPA deny messages in CI output | Fix metadata gaps or request policy exception | | Stale models not auto-retiring | Lifecycle cron job not running | Check CronJob status in Kubernetes | | Duplicate model versions | Race condition in CI pipeline | Add locking via registry API or database | | Missing eval evidence | Eval pipeline skipped or failed | Re-run eval suite and re-register | ## Related Skills - sbom-supply-chain (`sbom-supply-chain`) - Provenance and signing - policy-as-code (`policy-as-code`) - Enforce governance with policy engines - llm-fine-tuning (`llm-fine-tuning`) - Version adapters and training outputs - llmops-platform-engineering (`llmops-platform-engineering`) - Platform CI/CD and promotion workflows - ai-sre-incident-response (`ai-sre-incident-response`) - Incident response for model issues ## Limitations - Guidance executes against real environments: confirm target, blast radius, and rollback plan before applying anything. - Never deploy to production without explicit approval. Docs-only import: upstream scripts and templates not bundled. ### Example ```bash git status && git diff --stat kubectl diff -f manifest.yaml ``` > Adapted from [BagelHole/DevOps-Security-Agent-Skills](https://github.com/BagelHole/DevOps-Security-Agent-Skills) (MIT); frontmatter, When to Use/Limitations, and safety boundaries added for upstream compliance. Docs-only import: helper scripts and templates not bundled.