import asyncio import re import hashlib from datetime import datetime from aiva.db_connector import memory # Assuming db_connector.py exists # Mock classes for demonstration, replace with actual implementations class FoundationGate: def __init__(self, memory_stores): self.memory = memory_stores def validate(self, output, worker_id, metadata): """Enforce P1, P4, P8""" checks = {} # P1: Cryptographic/Tamper Check # In MVP, we verify the output_hash matches the metadata hash current_hash = hashlib.sha256(output.encode()).hexdigest() checks["tamper_proof"] = current_hash == metadata.get("output_hash") checks["signature_valid"] = True # Placeholder for Ed25519 signing # P4: Audit Log Status (pre-check) # We ensure metadata contains a valid timestamp checks["audit_ready"] = "timestamp" in metadata # P8: Privacy/PII Detection checks["no_pii_exposure"] = not self._detect_pii(output) score = sum(checks.values()) / len(checks) return score, checks def _detect_pii(self, text): pii_patterns = [ r'\b\d{3}-\d{2}-\d{4}\b', # SSN r'\b\d{16}\b', # Credit card r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b', # Email r'\b\d{3}-\d{3}-\d{4}\b', # Phone ] for pattern in pii_patterns: if re.search(pattern, text, re.IGNORECASE): return True return False class IntelligenceGate: def __init__(self, memory_stores): self.memory = memory_stores def validate(self, output, task_description, worker_id, metadata): """Enforce P5, P6, P3""" checks = {} checks["relevance"] = True checks["consistency"] = True checks["completeness"] = True score = sum(checks.values()) / len(checks) return score, checks class RealtimeGate: def __init__(self, memory_stores): self.memory = memory_stores def validate(self, output, task_description, worker_id, metadata): """Enforce P2, P7, P9""" checks = {} checks["latency"] = True checks["security"] = True checks["trust"] = True score = sum(checks.values()) / len(checks) return score, checks # validation_gates.py class GateAlpha: # Input Validity def __init__(self, memory): self.memory = memory self.name = "Gate Alpha" async def validate(self, input_data, metadata): """Validates the input data quality.""" try: is_valid = self._check_source_data_quality(input_data, metadata) reason = "Input data is valid." if is_valid else "Input data is invalid." await self.memory.log_audit(self.name, "input_validation", {"passed": is_valid, "reason": reason, "metadata": metadata}) return is_valid, reason except Exception as e: reason = f"Gate Alpha validation failed: {e}" await self.memory.log_audit(self.name, "input_validation", {"passed": False, "reason": reason, "metadata": metadata}) return False, reason def _check_source_data_quality(self, input_data, metadata): """Placeholder for input data quality checks. Implement data validation logic here.""" # Example: Check if required fields are present if not isinstance(input_data, dict): return False required_fields = metadata.get("required_fields", []) # Get required fields from metadata for field in required_fields: if field not in input_data: return False return True class GateBeta: # Output Quality def __init__(self, memory): self.memory = memory self.name = "Gate Beta" async def validate(self, output, task_description, metadata): """Checks accuracy and completeness of the output.""" try: is_accurate = self._check_accuracy(output, task_description, metadata) is_complete = self._check_completeness(output, task_description, metadata) passed = is_accurate and is_complete reason = "Output is accurate and complete." if passed else "Output is not accurate or complete." await self.memory.log_audit(self.name, "output_validation", {"passed": passed, "reason": reason, "metadata": metadata}) return passed, reason except Exception as e: reason = f"Gate Beta validation failed: {e}" await self.memory.log_audit(self.name, "output_validation", {"passed": False, "reason": reason, "metadata": metadata}) return False, reason def _check_accuracy(self, output, task_description, metadata): """Placeholder for accuracy checks. Implement accuracy validation logic here.""" # Example: Check if the output contains specific keywords required_keywords = metadata.get("required_keywords", []) if not isinstance(output, str): return False for keyword in required_keywords: if keyword not in output.lower(): return False return True def _check_completeness(self, output, task_description, metadata): """Placeholder for completeness checks. Implement completeness validation logic here.""" # Example: check output length min_length = metadata.get("min_length", 0) if not isinstance(output, str): return False return len(output) >= min_length class GateGamma: # Insight Purity def __init__(self, memory): self.memory = memory self.name = "Gate Gamma" async def validate(self, output, task_description, metadata): """Detects hallucinations in the output.""" try: no_hallucinations = not self._detect_hallucinations(output, task_description, metadata) reason = "No hallucinations detected." if no_hallucinations else "Hallucinations detected." await self.memory.log_audit(self.name, "hallucination_detection", {"passed": no_hallucinations, "reason": reason, "metadata": metadata}) return no_hallucinations, reason except Exception as e: reason = f"Gate Gamma validation failed: {e}" await self.memory.log_audit(self.name, "hallucination_detection", {"passed": False, "reason": reason, "metadata": metadata}) return False, reason def _detect_hallucinations(self, output, task_description, metadata): """Placeholder for hallucination detection. Implement hallucination detection logic here.""" # Example: Check for inconsistencies with known facts or the task description known_facts = metadata.get("known_facts", []) if not isinstance(output, str): return True for fact in known_facts: if fact.lower() not in output.lower(): return True # Assume hallucination if known fact is missing return False class GateDelta: # Memory Integration def __init__(self, memory): self.memory = memory self.name = "Gate Delta" async def validate(self, data, key, operation_type, metadata): """Validates memory storage operations.""" try: success = await self._validate_storage_operation(data, key, operation_type, metadata) reason = "Memory operation validated." if success else "Memory operation failed." await self.memory.log_audit(self.name, "memory_validation", {"passed": success, "reason": reason, "metadata": metadata, "operation": operation_type, "key": key}) return success, reason except Exception as e: reason = f"Gate Delta validation failed: {e}" await self.memory.log_audit(self.name, "memory_validation", {"passed": False, "reason": reason, "metadata": metadata, "operation": operation_type, "key": key}) return False, reason async def _validate_storage_operation(self, data, key, operation_type, metadata): """Placeholder for validating memory storage operations.""" # Example: Check if data was successfully written to memory if operation_type == "write": try: await self.memory.write_data(key, data) read_back = await self.memory.read_data(key) return read_back == data except Exception as e: print(f"Memory write/read failed: {e}") return False elif operation_type == "read": try: read_data = await self.memory.read_data(key) return read_data is not None except Exception as e: print(f"Memory read failed: {e}") return False return False class GateEpsilon: # Strategy Alignment def __init__(self, memory): self.memory = memory self.name = "Gate Epsilon" async def validate(self, output, task_description, metadata): """Ensures the output aligns with the revenue pathway.""" try: is_aligned = self._check_revenue_alignment(output, task_description, metadata) reason = "Output aligns with revenue pathway." if is_aligned else "Output does not align with revenue pathway." await self.memory.log_audit(self.name, "strategy_validation", {"passed": is_aligned, "reason": reason, "metadata": metadata}) return is_aligned, reason except Exception as e: reason = f"Gate Epsilon validation failed: {e}" await self.memory.log_audit(self.name, "strategy_validation", {"passed": False, "reason": reason, "metadata": metadata}) return False, reason def _check_revenue_alignment(self, output, task_description, metadata): """Placeholder for revenue alignment checks.""" # Example: Check if output promotes specific products or services promoted_products = metadata.get("promoted_products", []) if not isinstance(output, str): return False for product in promoted_products: if product.lower() in output.lower(): return True # Assume alignment if product is mentioned return False class GateZeta: # Budget Compliance def __init__(self, memory): self.memory = memory self.name = "Gate Zeta" async def validate(self, resource_usage, metadata): """Monitors resource usage and ensures budget compliance.""" try: is_compliant = self._check_budget_compliance(resource_usage, metadata) reason = "Resource usage is within budget." if is_compliant else "Resource usage exceeds budget." await self.memory.log_audit(self.name, "budget_validation", {"passed": is_compliant, "reason": reason, "metadata": metadata, "resource_usage": resource_usage}) return is_compliant, reason except Exception as e: reason = f"Gate Zeta validation failed: {e}" await self.memory.log_audit(self.name, "budget_validation", {"passed": False, "reason": reason, "metadata": metadata, "resource_usage": resource_usage}) return False, reason def _check_budget_compliance(self, resource_usage, metadata): """Placeholder for budget compliance checks.""" # Example: Check if CPU usage is below the threshold cpu_threshold = metadata.get("cpu_threshold", 0.8) if not isinstance(resource_usage, dict): return False cpu_usage = resource_usage.get("cpu_usage", 0.0) return cpu_usage <= cpu_threshold # Triple-Gate Validator Orchestrator (v2.0) # The CNS of Genesis Patent Validation class TripleGateValidator: def __init__(self): self.memory = memory self.gate1 = FoundationGate(memory) self.gate2 = IntelligenceGate(memory) self.gate3 = RealtimeGate(memory) self.gate_alpha = GateAlpha(memory) self.gate_beta = GateBeta(memory) self.gate_gamma = GateGamma(memory) self.gate_delta = GateDelta(memory) self.gate_epsilon = GateEpsilon(memory) self.gate_zeta = GateZeta(memory) async def validate_worker_output(self, input_data, output, task_description, worker_id, metadata, resource_usage): """Run complete validation suite""" # Gate Alpha: Input Validation alpha_passed, alpha_reason = await self.gate_alpha.validate(input_data, metadata) if not alpha_passed: return self._generate_report(worker_id, False, "Input Validation Failed", {"alpha": {"passed": alpha_passed, "reason": alpha_reason}}) # Gate Beta: Output Quality beta_passed, beta_reason = await self.gate_beta.validate(output, task_description, metadata) if not beta_passed: return self._generate_report(worker_id, False, "Output Quality Failed", {"beta": {"passed": beta_passed, "reason": beta_reason}}) # Gate Gamma: Insight Purity gamma_passed, gamma_reason = await self.gate_gamma.validate(output, task_description, metadata) if not gamma_passed: return self._generate_report(worker_id, False, "Insight Purity Failed", {"gamma": {"passed": gamma_passed, "reason": gamma_reason}}) # Gate Delta: Memory Integration (Example: Writing data to memory) delta_passed, delta_reason = await self.gate_delta.validate(output, worker_id, "write", metadata) # Using output as data and worker_id as key if not delta_passed: return self._generate_report(worker_id, False, "Memory Integration Failed", {"delta": {"passed": delta_passed, "reason": delta_reason}}) # Gate Epsilon: Strategy Alignment epsilon_passed, epsilon_reason = await self.gate_epsilon.validate(output, task_description, metadata) if not epsilon_passed: return self._generate_report(worker_id, False, "Strategy Alignment Failed", {"epsilon": {"passed": epsilon_passed, "reason": epsilon_reason}}) # Gate Zeta: Budget Compliance zeta_passed, zeta_reason = await self.gate_zeta.validate(resource_usage, metadata) if not zeta_passed: return self._generate_report(worker_id, False, "Budget Compliance Failed", {"zeta": {"passed": zeta_passed, "reason": zeta_reason}}) # Gate 1: Foundation (P1, P4, P8) g1_score, g1_checks = self.gate1.validate(output, worker_id, metadata) # Gate 2: Intelligence (P5, P6, P3) g2_score, g2_checks = self.gate2.validate(output, task_description, worker_id, metadata) # Gate 3: Real-Time (P2, P7, P9) g3_score, g3_checks = self.gate3.validate(output, task_description, worker_id, metadata) # Calculate overall score overall_score = (g1_score + g2_score + g3_score) / 3 passed = overall_score >= 0.8 report = { "timestamp": datetime.utcnow().isoformat(), "worker_id": worker_id, "passed": passed, "overall_score": overall_score, "gates": { "foundation": {"score": g1_score, "checks": g1_checks}, "intelligence": {"score": g2_score, "checks": g2_checks}, "realtime": {"score": g3_score, "checks": g3_checks}, "alpha": {"passed": alpha_passed, "reason": alpha_reason}, "beta": {"passed": beta_passed, "reason": beta_reason}, "gamma": {"passed": gamma_passed, "reason": gamma_reason}, "delta": {"passed": delta_passed, "reason": delta_reason}, "epsilon": {"passed": epsilon_passed, "reason": epsilon_reason}, "zeta": {"passed": zeta_passed, "reason": zeta_reason} } } # Log to immutable audit trail (Patent P4) try: await self.memory.log_audit("triple_gate_validation", worker_id, report) except Exception as e: print(f"[WARNING] Audit logging failed: {e}") # Update worker trust (Patent P9) try: await self.memory.update_worker_trust(worker_id, passed) except Exception as e: print(f"[WARNING] Trust update failed: {e}") return report def _generate_report(self, worker_id, passed, message, gate_results): report = { "timestamp": datetime.utcnow().isoformat(), "worker_id": worker_id, "passed": passed, "message": message, "gates": gate_results } # Log to immutable audit trail (Patent P4) try: asyncio.run(self.memory.log_audit("triple_gate_validation", worker_id, report)) except Exception as e: print(f"[WARNING] Audit logging failed: {e}") # Update worker trust (Patent P9) try: asyncio.run(self.memory.update_worker_trust(worker_id, passed)) except Exception as e: print(f"[WARNING] Trust update failed: {e}") return report # Global validator instance validator = TripleGateValidator()