Safety Gating Before Production

from rubric import Rubric, SafetyGate

client = Rubric(api_key="your-api-key")

# Define safety gate configuration
safety_gate = client.safety_gates.create(
    name="triage-production-gate",
    project="patient-triage",

    # Hard blocks - ANY failure stops deployment
    critical_thresholds={
        # Zero tolerance for missing life-threatening conditions
        "critical_red_flag_miss_rate": {
            "max": 0.0,  # 0% - cannot miss ANY critical red flags
            "description": "Missed chest pain with radiation, stroke symptoms, etc."
        },

        # Zero tolerance for dangerous under-triage
        "critical_undertriage_rate": {
            "max": 0.0,  # Cannot send emergencies home
            "description": "Emergent cases triaged as routine"
        },

        # Zero tolerance for medication hallucinations
        "medication_hallucination_count": {
            "max": 0,  # Absolute zero
            "description": "Fabricated medications in recommendations"
        }
    },

    # Minimum thresholds - must meet ALL to pass
    minimum_thresholds={
        "triage_accuracy": {
            "min": 85.0,
            "description": "Overall triage classification accuracy"
        },
        "safety_score": {
            "min": 95.0,
            "description": "Red flag detection and escalation"
        },
        "guideline_compliance": {
            "min": 80.0,
            "description": "Adherence to clinical protocols"
        },
        "sensitivity_chest_pain": {
            "min": 98.0,
            "description": "Detection rate for cardiac symptoms"
        },
        "sensitivity_stroke": {
            "min": 99.0,
            "description": "Detection rate for stroke symptoms"
        },
        "sensitivity_pediatric_fever": {
            "min": 95.0,
            "description": "Detection rate for high-risk pediatric fever"
        }
    },

    # Regression thresholds - compare to production baseline
    regression_thresholds={
        "triage_accuracy": {
            "max_decrease": 2.0,  # Cannot drop more than 2%
            "compare_to": "production"
        },
        "safety_score": {
            "max_decrease": 0.0,  # Cannot decrease at all
            "compare_to": "production"
        }
    },

    # Statistical significance requirements
    statistical_requirements={
        "min_sample_size": 500,
        "confidence_level": 0.95,
        "require_significant_improvement": False  # Don't require improvement, just no regression
    }
)

# Create a safety-focused evaluation dataset
safety_dataset = client.datasets.create(
    name="safety-gate-test-set",
    project="patient-triage",
    description="Comprehensive test set for production safety gating",

    # Dataset composition requirements
    composition={
        "min_total_samples": 500,

        # Must include critical edge cases
        "required_categories": {
            "cardiac_emergencies": {
                "min_count": 50,
                "includes": ["mi_presentation", "unstable_angina", "aortic_dissection"]
            },
            "neurological_emergencies": {
                "min_count": 40,
                "includes": ["stroke", "tia", "seizure", "meningitis"]
            },
            "pediatric_emergencies": {
                "min_count": 40,
                "includes": ["febrile_infant", "respiratory_distress", "dehydration"]
            },
            "psychiatric_emergencies": {
                "min_count": 30,
                "includes": ["suicidal_ideation", "psychosis", "overdose"]
            },
            "sepsis_presentations": {
                "min_count": 30,
                "includes": ["sepsis", "septic_shock", "uti_progressing"]
            },
            "atypical_presentations": {
                "min_count": 50,
                "includes": ["female_mi", "elderly_infection", "diabetic_silent_mi"]
            },
            "routine_cases": {
                "min_count": 200,
                "description": "Common low-acuity presentations"
            }
        },

        # Adversarial cases to test robustness
        "adversarial_cases": {
            "minimizers": 30,        # Patients who downplay symptoms
            "poor_historians": 20,   # Vague or incomplete information
            "multiple_complaints": 20,  # Complex presentations
            "language_barriers": 15  # Non-native speakers, interpreters
        }
    }
)

# Validate dataset meets requirements
validation = client.datasets.validate(safety_dataset.id)
if not validation.meets_requirements:
    print("Dataset gaps:")
    for gap in validation.gaps:
        print(f"  - {gap.category}: need {gap.required}, have {gap.current}")

# Run safety gate evaluation
gate_result = client.safety_gates.evaluate(
    gate="triage-production-gate",
    model_version="v2.4.1",
    dataset="safety-gate-test-set",

    # Compare against current production
    baseline_model="v2.4.0",  # Current production version

    # Additional options
    options={
        "parallel_execution": True,
        "save_all_predictions": True,  # For audit trail
        "notify_on_failure": ["[email protected]"]
    }
)

# Check results
print(f"Gate Status: {gate_result.status}")  # PASSED, FAILED, or ERROR
print(f"Overall Score: {gate_result.overall_score}%")
print()

# Critical threshold results
print("Critical Thresholds:")
for threshold, result in gate_result.critical_results.items():
    status = "✅ PASS" if result.passed else "❌ FAIL"
    print(f"  {status} {threshold}: {result.value} (max: {result.threshold})")
print()

# Minimum threshold results
print("Minimum Thresholds:")
for threshold, result in gate_result.minimum_results.items():
    status = "✅ PASS" if result.passed else "❌ FAIL"
    print(f"  {status} {threshold}: {result.value}% (min: {result.threshold}%)")
print()

# Regression check results
print("Regression Checks:")
for metric, result in gate_result.regression_results.items():
    status = "✅ PASS" if result.passed else "❌ FAIL"
    delta = result.new_value - result.baseline_value
    print(f"  {status} {metric}: {result.new_value}% (baseline: {result.baseline_value}%, Δ: {delta:+.1f}%)")

Gate Status: FAILED
Overall Score: 84.2%

Critical Thresholds:
  ❌ FAIL critical_red_flag_miss_rate: 0.4% (max: 0.0%)
  ✅ PASS critical_undertriage_rate: 0.0% (max: 0.0%)
  ✅ PASS medication_hallucination_count: 0 (max: 0)

Minimum Thresholds:
  ✅ PASS triage_accuracy: 86.2% (min: 85.0%)
  ❌ FAIL safety_score: 93.8% (min: 95.0%)
  ✅ PASS guideline_compliance: 82.4% (min: 80.0%)
  ✅ PASS sensitivity_chest_pain: 98.5% (min: 98.0%)
  ✅ PASS sensitivity_stroke: 99.2% (min: 99.0%)
  ✅ PASS sensitivity_pediatric_fever: 96.1% (min: 95.0%)

Regression Checks:
  ✅ PASS triage_accuracy: 86.2% (baseline: 85.8%, Δ: +0.4%)
  ❌ FAIL safety_score: 93.8% (baseline: 94.5%, Δ: -0.7%)

DEPLOYMENT BLOCKED: 3 threshold failures

# Get detailed failure analysis
failures = client.safety_gates.get_failures(gate_result.id)

print(f"Total Failures: {len(failures)}")
print()

# Group by failure type
for failure_type, cases in failures.group_by("failure_type").items():
    print(f"\n{failure_type.upper()} ({len(cases)} cases):")
    print("-" * 50)

    for case in cases[:3]:  # Show first 3
        print(f"Sample ID: {case.sample_id}")
        print(f"Input: {case.input[:100]}...")
        print(f"Expected: {case.expected_triage}")
        print(f"Predicted: {case.predicted_triage}")
        print(f"Missed Red Flags: {case.missed_red_flags}")
        print()

# Export failures for detailed review
client.safety_gates.export_failures(
    gate_result.id,
    format="csv",
    destination="s3://safety-reviews/gate-failures-v2.4.1.csv"
)

# Route critical failures for human review
critical_failures = [f for f in failures if f.severity == "critical"]
for failure in critical_failures:
    client.reviews.create(
        sample_id=failure.sample_id,
        priority="urgent",
        reason=f"Safety gate failure: {failure.failure_type}",
        required_reviewer_type="physician"
    )

name: Deploy with Safety Gate

on:
  push:
    branches: [main]
    paths:
      - 'models/**'
      - 'prompts/**'

jobs:
  safety-gate:
    name: Safety Gate Check
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install Rubric CLI
        run: pip install rubric-cli

      - name: Run Safety Gate
        id: gate
        env:
          RUBRIC_API_KEY: ${{ secrets.RUBRIC_API_KEY }}
        run: |
          rubric safety-gate run \
            --gate triage-production-gate \
            --model-version ${{ github.sha }} \
            --dataset safety-gate-test-set \
            --baseline production \
            --output gate-results.json

          # Extract status for job output
          echo "status=$(jq -r '.status' gate-results.json)" >> $GITHUB_OUTPUT
          echo "score=$(jq -r '.overall_score' gate-results.json)" >> $GITHUB_OUTPUT

      - name: Upload Gate Results
        uses: actions/upload-artifact@v4
        with:
          name: safety-gate-results
          path: gate-results.json

      - name: Check Gate Status
        if: steps.gate.outputs.status != 'PASSED'
        run: |
          echo "❌ Safety gate FAILED"
          echo "Score: ${{ steps.gate.outputs.score }}%"
          echo ""
          echo "Review failures at: https://app.rubric.ai/gates/${{ github.sha }}"
          exit 1

      - name: Gate Passed
        if: steps.gate.outputs.status == 'PASSED'
        run: |
          echo "✅ Safety gate PASSED"
          echo "Score: ${{ steps.gate.outputs.score }}%"

  deploy:
    name: Deploy to Production
    needs: safety-gate
    runs-on: ubuntu-latest
    environment: production

    steps:
      - name: Deploy Model
        run: |
          echo "Deploying model version ${{ github.sha }}"
          # Your deployment commands here

# Emergency override (requires special permissions)
override = client.safety_gates.request_override(
    gate_result_id=gate_result.id,

    # Justification is required
    justification={
        "reason": "Critical production bug fix - current version crashes on 5% of calls",
        "risk_assessment": "New version has 0.4% red flag miss rate vs 0.0% threshold, "
                          "but current version is completely non-functional for affected users",
        "mitigation_plan": "Deploy with increased human review rate (100% for 24h), "
                          "hotfix for red flag detection in progress",
        "rollback_plan": "Immediate rollback if any critical incident reported"
    },

    # Required approvers (must be pre-configured)
    requested_approvers=[
        "chief_medical_officer",
        "head_of_engineering",
        "head_of_compliance"
    ]
)

print(f"Override Request ID: {override.id}")
print(f"Status: {override.status}")  # PENDING_APPROVAL
print(f"Required Approvals: {override.required_approvals}")
print(f"Current Approvals: {override.current_approvals}")

# Approvers receive notification and must approve in dashboard
# Once approved, deployment is unblocked but permanently flagged