import asyncio import re import hashlib from datetime import datetime from aiva.db_connector import memory # Mock DB connector (replace with actual implementation) # Assume memory object has methods: # - log_audit(gate_name, worker_id, report) # - update_worker_trust(worker_id, passed) # - get_source_data(source_id) # - store_data(key, value) # - retrieve_data(key) class GateBase: def __init__(self, memory): self.memory = memory async def validate(self, output, task_description, worker_id, metadata): raise NotImplementedError async def _log_decision(self, gate_name, worker_id, passed, reason, metadata): report = { "timestamp": datetime.utcnow().isoformat(), "worker_id": worker_id, "passed": passed, "reason": reason, "metadata": metadata } try: await asyncio.sleep(0.001) # Simulate async op self.memory.log_audit(gate_name, worker_id, report) except Exception as e: print(f"[WARNING] Audit logging failed for {gate_name}: {e}") class GateAlpha(GateBase): """Gate Alpha: Input Validity (Source Data Quality)""" async def validate(self, output, task_description, worker_id, metadata): source_id = metadata.get("source_id") if not source_id: await self._log_decision("GateAlpha", worker_id, False, "Missing source_id in metadata", metadata) return False, "Missing source_id" try: await asyncio.sleep(0.001) # Simulate async op source_data = self.memory.get_source_data(source_id) if not source_data: await self._log_decision("GateAlpha", worker_id, False, f"Source data not found for id: {source_id}", metadata) return False, "Source data not found" # Example: Check if source data is too short if len(source_data) < 10: await self._log_decision("GateAlpha", worker_id, False, "Source data too short", metadata) return False, "Source data too short" await self._log_decision("GateAlpha", worker_id, True, "Source data valid", metadata) return True, "Source data valid" except Exception as e: await self._log_decision("GateAlpha", worker_id, False, f"Error validating source data: {e}", metadata) return False, f"Error validating source data: {e}" class GateBeta(GateBase): """Gate Beta: Output Quality (Accuracy and Completeness)""" async def validate(self, output, task_description, worker_id, metadata): # Example: Check if output contains specific keywords from task description required_keywords = re.findall(r'\b\w+\b', task_description, re.IGNORECASE) missing_keywords = [keyword for keyword in required_keywords if keyword.lower() not in output.lower()] if missing_keywords: reason = f"Output missing keywords: {', '.join(missing_keywords)}" await self._log_decision("GateBeta", worker_id, False, reason, metadata) return False, reason # Example: Check if output length is within acceptable range min_length = metadata.get("min_length", 50) max_length = metadata.get("max_length", 500) output_length = len(output) if output_length < min_length or output_length > max_length: reason = f"Output length ({output_length}) outside acceptable range ({min_length}-{max_length})" await self._log_decision("GateBeta", worker_id, False, reason, metadata) return False, reason await self._log_decision("GateBeta", worker_id, True, "Output quality is good", metadata) return True, "Output quality is good" class GateGamma(GateBase): """Gate Gamma: Insight Purity (Detect Hallucinations)""" async def validate(self, output, task_description, worker_id, metadata): # Example: Check for nonsensical phrases or contradictions # (This requires a more sophisticated NLP model in a real-world scenario) nonsense_patterns = [ r"the sky is green", r"water is dry", ] for pattern in nonsense_patterns: if re.search(pattern, output, re.IGNORECASE): reason = f"Detected nonsensical phrase: {pattern}" await self._log_decision("GateGamma", worker_id, False, reason, metadata) return False, reason # Example: Check if output contradicts task description if "don't mention cats" in task_description.lower() and "cat" in output.lower(): reason = "Output contradicts task description (mentions cats)" await self._log_decision("GateGamma", worker_id, False, reason, metadata) return False, reason await self._log_decision("GateGamma", worker_id, True, "No hallucinations detected", metadata) return True, "No hallucinations detected" class GateDelta(GateBase): """Gate Delta: Memory Integration (Validate Storage Operations)""" async def validate(self, output, task_description, worker_id, metadata): key = metadata.get("storage_key") if not key: await self._log_decision("GateDelta", worker_id, False, "Missing storage_key in metadata", metadata) return False, "Missing storage_key" try: # Simulate storing data await asyncio.sleep(0.001) # Simulate async op self.memory.store_data(key, output) # Simulate retrieving data retrieved_data = self.memory.retrieve_data(key) if retrieved_data != output: reason = "Data storage/retrieval mismatch" await self._log_decision("GateDelta", worker_id, False, reason, metadata) return False, reason await self._log_decision("GateDelta", worker_id, True, "Memory integration successful", metadata) return True, "Memory integration successful" except Exception as e: reason = f"Memory integration failed: {e}" await self._log_decision("GateDelta", worker_id, False, reason, metadata) return False, reason class GateEpsilon(GateBase): """Gate Epsilon: Strategy Alignment (Ensure Revenue Pathway Fit)""" async def validate(self, output, task_description, worker_id, metadata): # This gate requires a deep understanding of the business strategy # and how the output contributes to revenue generation. # Example: Check if output aligns with target customer profile target_customer = metadata.get("target_customer") if target_customer and target_customer.lower() not in output.lower(): reason = f"Output does not mention target customer: {target_customer}" await self._log_decision("GateEpsilon", worker_id, False, reason, metadata) return False, reason # Example: Check if output promotes a specific product or service product_promotion = metadata.get("product_promotion") if product_promotion and product_promotion.lower() not in output.lower(): reason = f"Output does not promote product: {product_promotion}" await self._log_decision("GateEpsilon", worker_id, False, reason, metadata) return False, reason await self._log_decision("GateEpsilon", worker_id, True, "Output aligns with business strategy", metadata) return True, "Output aligns with business strategy" class GateZeta(GateBase): """Gate Zeta: Budget Compliance (Resource Monitoring)""" async def validate(self, output, task_description, worker_id, metadata): # This gate requires access to resource usage data (e.g., API calls, compute time) # Assume metadata contains estimated cost of generating the output estimated_cost = metadata.get("estimated_cost") budget_limit = metadata.get("budget_limit", 1.0) # Default budget limit of 1.0 if estimated_cost is None: await self._log_decision("GateZeta", worker_id, False, "Missing estimated_cost in metadata", metadata) return False, "Missing estimated_cost" if estimated_cost > budget_limit: reason = f"Estimated cost ({estimated_cost}) exceeds budget limit ({budget_limit})" await self._log_decision("GateZeta", worker_id, False, reason, metadata) return False, reason await self._log_decision("GateZeta", worker_id, True, "Output within budget", metadata) return True, "Output within budget" class SixGateValidator: def __init__(self, memory): self.memory = 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, output, task_description, worker_id, metadata): """Run complete 6-gate validation suite""" # Run gates sequentially, chaining on pass gate_results = {} alpha_passed, alpha_reason = await self.gate_alpha.validate(output, task_description, worker_id, metadata) gate_results["alpha"] = {"passed": alpha_passed, "reason": alpha_reason} if not alpha_passed: return False, gate_results # Stop if any gate fails beta_passed, beta_reason = await self.gate_beta.validate(output, task_description, worker_id, metadata) gate_results["beta"] = {"passed": beta_passed, "reason": beta_reason} if not beta_passed: return False, gate_results gamma_passed, gamma_reason = await self.gate_gamma.validate(output, task_description, worker_id, metadata) gate_results["gamma"] = {"passed": gamma_passed, "reason": gamma_reason} if not gamma_passed: return False, gate_results delta_passed, delta_reason = await self.gate_delta.validate(output, task_description, worker_id, metadata) gate_results["delta"] = {"passed": delta_passed, "reason": delta_reason} if not delta_passed: return False, gate_results epsilon_passed, epsilon_reason = await self.gate_epsilon.validate(output, task_description, worker_id, metadata) gate_results["epsilon"] = {"passed": epsilon_passed, "reason": epsilon_reason} if not epsilon_passed: return False, gate_results zeta_passed, zeta_reason = await self.gate_zeta.validate(output, task_description, worker_id, metadata) gate_results["zeta"] = {"passed": zeta_passed, "reason": zeta_reason} if not zeta_passed: return False, gate_results return True, gate_results # Example Usage (replace with actual memory implementation) class MockMemory: def __init__(self): self.audit_logs = [] self.worker_trust = {} self.source_data = {"source1": "This is some sample source data."} self.data_store = {} def log_audit(self, gate_name, worker_id, report): self.audit_logs.append({"gate": gate_name, "worker_id": worker_id, "report": report}) print(f"Audit Logged: {gate_name}, {worker_id}") def update_worker_trust(self, worker_id, passed): if worker_id not in self.worker_trust: self.worker_trust[worker_id] = {"successes": 0, "failures": 0} if passed: self.worker_trust[worker_id]["successes"] += 1 else: self.worker_trust[worker_id]["failures"] += 1 print(f"Worker Trust Updated: {worker_id}, Passed: {passed}") def get_source_data(self, source_id): return self.source_data.get(source_id) def store_data(self, key, value): self.data_store[key] = value print(f"Data Stored: key={key}") def retrieve_data(self, key): return self.data_store.get(key) async def main(): memory = MockMemory() validator = SixGateValidator(memory) output = "This is a sample output mentioning our target customer: premium users. We are promoting product A." task_description = "Generate a marketing message promoting product A." worker_id = "worker123" metadata = { "source_id": "source1", "storage_key": "output_key", "target_customer": "premium users", "product_promotion": "product A", "estimated_cost": 0.5, "budget_limit": 1.0 } passed, results = await validator.validate_worker_output(output, task_description, worker_id, metadata) print(f"Validation Passed: {passed}") print(f"Gate Results: {results}") if __name__ == "__main__": asyncio.run(main())