add compliance module with agent inventory, risk classification, and FREE-AI reporting

Author: shivdeep1

airlock/compliance/__init__.py

@@ -0,0 +1,31 @@
+"""RBI FREE-AI compliance module for the Airlock Protocol."""
+
+from __future__ import annotations
+
+from airlock.compliance.bias_detector import BiasDetector
+from airlock.compliance.free_ai_mapper import FreeAIMapper
+from airlock.compliance.incident import IncidentStore
+from airlock.compliance.inventory import AgentInventory
+from airlock.compliance.report_generator import ComplianceReportGenerator
+from airlock.compliance.risk_classifier import RiskClassifier
+from airlock.compliance.schemas import (
+    AgentInventoryEntry,
+    ComplianceReport,
+    IncidentReport,
+    RiskClassification,
+    RiskLevel,
+)
+
+__all__ = [
+    "AgentInventory",
+    "AgentInventoryEntry",
+    "BiasDetector",
+    "ComplianceReport",
+    "ComplianceReportGenerator",
+    "FreeAIMapper",
+    "IncidentReport",
+    "IncidentStore",
+    "RiskClassification",
+    "RiskClassifier",
+    "RiskLevel",
+]

airlock/compliance/bias_detector.py

@@ -0,0 +1,133 @@
+"""Basic bias detection for AI agent behavior patterns -- FREE-AI Rec #15."""
+
+from __future__ import annotations
+
+import logging
+import statistics
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class BiasDetector:
+    """Basic bias detection for AI agent behavior patterns."""
+
+    def analyze_verification_patterns(
+        self, verification_results: list[dict[str, Any]]
+    ) -> dict[str, Any]:
+        """Detect statistical bias in verification outcomes.
+
+        Looks for skewed pass/fail rates that could indicate systematic bias
+        in the verification process.
+        """
+        if not verification_results:
+            return {
+                "bias_detected": False,
+                "bias_type": "none",
+                "confidence": 0.0,
+                "details": "No verification data available",
+            }
+
+        total = len(verification_results)
+        passed = sum(1 for r in verification_results if r.get("result") == "passed")
+        failed = total - passed
+
+        pass_rate = passed / total if total > 0 else 0.0
+
+        # Check for extreme skew (>90% or <10% pass rate with sufficient data)
+        if total >= 10 and (pass_rate > 0.9 or pass_rate < 0.1):
+            bias_type = "high_pass_rate" if pass_rate > 0.9 else "high_fail_rate"
+            return {
+                "bias_detected": True,
+                "bias_type": bias_type,
+                "confidence": min(0.5 + (total / 100), 0.95),
+                "details": (
+                    f"Verification pass rate is {pass_rate:.1%} "
+                    f"across {total} verifications (passed={passed}, failed={failed})"
+                ),
+            }
+
+        return {
+            "bias_detected": False,
+            "bias_type": "none",
+            "confidence": min(0.5 + (total / 100), 0.95),
+            "details": (
+                f"Verification pass rate is {pass_rate:.1%} "
+                f"across {total} verifications -- within expected range"
+            ),
+        }
+
+    def analyze_trust_distribution(
+        self, trust_scores: list[float]
+    ) -> dict[str, Any]:
+        """Analyze trust score distribution for anomalies.
+
+        Checks for abnormal clustering, extreme values, or unexpected
+        distribution patterns that could indicate bias.
+        """
+        if not trust_scores:
+            return {
+                "anomaly_detected": False,
+                "distribution_type": "unknown",
+                "details": "No trust score data available",
+            }
+
+        if len(trust_scores) < 2:
+            return {
+                "anomaly_detected": False,
+                "distribution_type": "insufficient_data",
+                "mean": trust_scores[0] if trust_scores else 0.0,
+                "details": "Insufficient data for distribution analysis",
+            }
+
+        mean = statistics.mean(trust_scores)
+        stdev = statistics.stdev(trust_scores)
+        median = statistics.median(trust_scores)
+
+        # Detect clustering at extremes
+        low_cluster = sum(1 for s in trust_scores if s < 0.2) / len(trust_scores)
+        high_cluster = sum(1 for s in trust_scores if s > 0.8) / len(trust_scores)
+
+        anomaly = False
+        distribution_type = "normal"
+        details_parts: list[str] = []
+
+        if stdev < 0.05 and len(trust_scores) >= 5:
+            anomaly = True
+            distribution_type = "uniform_cluster"
+            details_parts.append(
+                f"Scores are unusually clustered (stdev={stdev:.3f})"
+            )
+
+        if low_cluster > 0.5:
+            anomaly = True
+            distribution_type = "low_skew"
+            details_parts.append(
+                f"{low_cluster:.0%} of scores below 0.2"
+            )
+
+        if high_cluster > 0.8:
+            anomaly = True
+            distribution_type = "high_skew"
+            details_parts.append(
+                f"{high_cluster:.0%} of scores above 0.8"
+            )
+
+        if abs(mean - median) > 0.2:
+            anomaly = True
+            distribution_type = "skewed"
+            details_parts.append(
+                f"Mean-median gap of {abs(mean - median):.2f} suggests skew"
+            )
+
+        if not details_parts:
+            details_parts.append("Distribution appears normal")
+
+        return {
+            "anomaly_detected": anomaly,
+            "distribution_type": distribution_type,
+            "mean": round(mean, 4),
+            "stdev": round(stdev, 4),
+            "median": round(median, 4),
+            "details": "; ".join(details_parts),
+        }

airlock/compliance/free_ai_mapper.py

@@ -0,0 +1,168 @@
+"""Maps Airlock data to RBI FREE-AI 7 Sutras and 26 recommendations."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from airlock.compliance.incident import IncidentStore
+from airlock.compliance.inventory import AgentInventory
+
+logger = logging.getLogger(__name__)
+
+# FREE-AI Sutras
+SUTRAS: dict[str, str] = {
+    "sutra_1": "Governance & Oversight",
+    "sutra_2": "Risk Management",
+    "sutra_3": "Data Privacy & Security",
+    "sutra_4": "Transparency & Explainability",
+    "sutra_5": "Fairness & Non-discrimination",
+    "sutra_6": "Accountability & Audit",
+    "sutra_7": "Consumer Protection",
+}
+
+# Recommendation -> Airlock feature mapping
+RECOMMENDATION_MAP: dict[str, dict[str, str]] = {
+    "rec_13": {
+        "title": "Incident Reporting",
+        "airlock_feature": "incident_tracking",
+        "sutra": "sutra_2",
+    },
+    "rec_14": {
+        "title": "AI Model Inventory",
+        "airlock_feature": "agent_inventory",
+        "sutra": "sutra_1",
+    },
+    "rec_15": {
+        "title": "Auditor Assessment by Risk Profile",
+        "airlock_feature": "risk_classification",
+        "sutra": "sutra_2",
+    },
+    "rec_16": {
+        "title": "AI Disclosures in Annual Reports",
+        "airlock_feature": "compliance_reports",
+        "sutra": "sutra_4",
+    },
+    "rec_19": {
+        "title": "Internal Audit Proportional to Risk",
+        "airlock_feature": "audit_trail",
+        "sutra": "sutra_6",
+    },
+    "rec_20": {
+        "title": "Red Teaming",
+        "airlock_feature": "adversarial_testing",
+        "sutra": "sutra_2",
+    },
+}
+
+
+class FreeAIMapper:
+    """Map Airlock compliance state to FREE-AI recommendations."""
+
+    def map_compliance_status(
+        self,
+        inventory: AgentInventory,
+        incident_store: IncidentStore,
+    ) -> dict[str, dict[str, Any]]:
+        """Map current Airlock state to FREE-AI recommendation compliance status.
+
+        Returns a dict keyed by recommendation ID with compliance details.
+        """
+        result: dict[str, dict[str, Any]] = {}
+        for rec_id, rec_info in RECOMMENDATION_MAP.items():
+            result[rec_id] = self.get_recommendation_status(
+                rec_id, inventory, incident_store
+            )
+        return result
+
+    def get_recommendation_status(
+        self,
+        rec_id: str,
+        inventory: AgentInventory,
+        incident_store: IncidentStore,
+    ) -> dict[str, Any]:
+        """Get compliance status for a specific recommendation."""
+        rec_info = RECOMMENDATION_MAP.get(rec_id)
+        if rec_info is None:
+            return {
+                "status": "unknown",
+                "title": "Unknown Recommendation",
+                "details": f"Recommendation {rec_id} not mapped",
+            }
+
+        feature = rec_info["airlock_feature"]
+        sutra = rec_info["sutra"]
+        title = rec_info["title"]
+
+        status = self._assess_feature_status(feature, inventory, incident_store)
+
+        return {
+            "rec_id": rec_id,
+            "title": title,
+            "sutra": sutra,
+            "sutra_name": SUTRAS.get(sutra, "Unknown"),
+            "airlock_feature": feature,
+            "status": status,
+        }
+
+    def get_sutra_summary(
+        self,
+        inventory: AgentInventory,
+        incident_store: IncidentStore,
+    ) -> dict[str, dict[str, Any]]:
+        """Get compliance summary grouped by Sutra."""
+        full_mapping = self.map_compliance_status(inventory, incident_store)
+
+        sutra_summary: dict[str, dict[str, Any]] = {}
+        for sutra_id, sutra_name in SUTRAS.items():
+            recs = [
+                v
+                for v in full_mapping.values()
+                if v.get("sutra") == sutra_id
+            ]
+            compliant = sum(1 for r in recs if r.get("status") == "compliant")
+            total = len(recs)
+            sutra_summary[sutra_id] = {
+                "name": sutra_name,
+                "recommendations": total,
+                "compliant": compliant,
+                "status": "compliant" if compliant == total and total > 0 else "partial",
+            }
+        return sutra_summary
+
+    def _assess_feature_status(
+        self,
+        feature: str,
+        inventory: AgentInventory,
+        incident_store: IncidentStore,
+    ) -> str:
+        """Assess whether a specific Airlock feature meets compliance."""
+        if feature == "agent_inventory":
+            # Compliant if at least one agent is registered
+            return "compliant" if len(inventory) > 0 else "not_implemented"
+
+        if feature == "risk_classification":
+            # Compliant if agents have been assessed
+            entries = inventory.list_all()
+            if not entries:
+                return "not_implemented"
+            assessed = sum(1 for e in entries if e.last_assessed_at is not None)
+            return "compliant" if assessed > 0 else "partial"
+
+        if feature == "incident_tracking":
+            # Feature is available (store exists), compliant
+            return "compliant"
+
+        if feature == "compliance_reports":
+            # Feature is available (generator exists), compliant
+            return "compliant"
+
+        if feature == "audit_trail":
+            # Audit trail is always available in Airlock
+            return "compliant"
+
+        if feature == "adversarial_testing":
+            # Red teaming is a process, not a feature -- mark as partial
+            return "partial"
+
+        return "unknown"