import asyncio import re import hashlib from datetime import datetime from aiva.db_connector import memory # Assuming this exists # --- Generic Gate Base Class --- class ValidationGate: def __init__(self, name, memory): self.name = name self.memory = memory self.next_gate = None # For chaining def set_next(self, gate): self.next_gate = gate return gate # For convenient chaining async def validate(self, output, task_description, worker_id, metadata): raise NotImplementedError("Subclasses must implement the validate method") async def _log_decision(self, worker_id, passed, reason, data=None): log_entry = { "gate": self.name, "timestamp": datetime.utcnow().isoformat(), "worker_id": worker_id, "passed": passed, "reason": reason, "data": data or {} } try: self.memory.log_gate_decision(log_entry) # Assuming this method exists except Exception as e: print(f"[WARNING] Gate {self.name} logging failed: {e}") async def process_next(self, output, task_description, worker_id, metadata): if self.next_gate: return await self.next_gate.validate(output, task_description, worker_id, metadata) else: return True, "Validation chain complete" # End of chain # --- Specific Gate Implementations --- class GateAlpha(ValidationGate): def __init__(self, memory): super().__init__("Alpha", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Alpha: Input validity - verify source data quality.""" data_quality = metadata.get("data_quality", 0.0) # Assuming data_quality is in metadata passed = data_quality >= 0.7 # Example threshold if passed: reason = "Source data quality is acceptable." else: reason = f"Source data quality is too low ({data_quality})." await self._log_decision(worker_id, passed, reason, {"data_quality": data_quality}) if passed: return await self.process_next(output, task_description, worker_id, metadata) else: return False, reason class GateBeta(ValidationGate): def __init__(self, memory): super().__init__("Beta", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Beta: Output quality - check accuracy and completeness.""" accuracy_score = metadata.get("accuracy_score", 0.0) # Assuming accuracy score exists completeness_score = metadata.get("completeness_score", 0.0) passed = accuracy_score >= 0.8 and completeness_score >= 0.8 if passed: reason = "Output accuracy and completeness are acceptable." else: reason = f"Output quality too low (accuracy: {accuracy_score}, completeness: {completeness_score})." await self._log_decision(worker_id, passed, reason, {"accuracy": accuracy_score, "completeness": completeness_score}) if passed: return await self.process_next(output, task_description, worker_id, metadata) else: return False, reason class GateGamma(ValidationGate): def __init__(self, memory): super().__init__("Gamma", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Gamma: Insight purity - detect hallucinations.""" hallucination_score = self._detect_hallucinations(output, task_description) passed = hallucination_score <= 0.2 # Example: Lower score = less hallucinations if passed: reason = "Hallucination level is acceptable." else: reason = f"High hallucination level detected ({hallucination_score})." await self._log_decision(worker_id, passed, reason, {"hallucination_score": hallucination_score}) if passed: return await self.process_next(output, task_description, worker_id, metadata) else: return False, reason def _detect_hallucinations(self, output, task_description): """Dummy hallucination detection. Replace with real logic.""" # Simple example: Penalize if the output contains the word "unicorn" if "unicorn" in output.lower(): return 0.9 # High hallucination score else: return 0.1 class GateDelta(ValidationGate): def __init__(self, memory): super().__init__("Delta", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Delta: Memory integration - validate storage operations.""" try: # Simulate a memory write operation self.memory.write_data(worker_id, output) reason = "Memory write successful." passed = True except Exception as e: reason = f"Memory write failed: {e}" passed = False await self._log_decision(worker_id, passed, reason) if passed: return await self.process_next(output, task_description, worker_id, metadata) else: return False, reason class GateEpsilon(ValidationGate): def __init__(self, memory): super().__init__("Epsilon", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Epsilon: Strategy alignment - ensure revenue pathway fit.""" revenue_potential = metadata.get("revenue_potential", 0.0) passed = revenue_potential >= 0.5 # Example threshold if passed: reason = "Output aligns with revenue generation strategy." else: reason = f"Output does not sufficiently align with revenue strategy ({revenue_potential})." await self._log_decision(worker_id, passed, reason, {"revenue_potential": revenue_potential}) if passed: return await self.process_next(output, task_description, worker_id, metadata) else: return False, reason class GateZeta(ValidationGate): def __init__(self, memory): super().__init__("Zeta", memory) async def validate(self, output, task_description, worker_id, metadata): """Gate Zeta: Budget compliance - resource monitoring.""" cost = metadata.get("cost", 0.0) budget = metadata.get("budget", 100.0) # Example budget passed = cost <= budget if passed: reason = "Cost is within budget." else: reason = f"Cost exceeds budget (cost: {cost}, budget: {budget})." await self._log_decision(worker_id, passed, reason, {"cost": cost, "budget": budget}) return passed, reason # Last gate, no need to process next # --- Orchestrator Class --- class SixGateValidator: def __init__(self, memory): self.memory = memory # Create the gates self.alpha = GateAlpha(memory) self.beta = GateBeta(memory) self.gamma = GateGamma(memory) self.delta = GateDelta(memory) self.epsilon = GateEpsilon(memory) self.zeta = GateZeta(memory) # Chain the gates self.alpha.set_next(self.beta).set_next(self.gamma).set_next(self.delta).set_next(self.epsilon).set_next(self.zeta) async def validate_worker_output(self, output, task_description, worker_id, metadata): """Run complete 6-gate validation suite.""" return await self.alpha.validate(output, task_description, worker_id, metadata) # Example usage (assuming 'memory' is an instance of your memory store): # validator = SixGateValidator(memory) # passed, reason = await validator.validate_worker_output(output, task_description, worker_id, metadata)