import asyncio import re import hashlib from datetime import datetime from aiva.db_connector import memory # Assuming this exists # --- Gate Definitions --- class ValidationGate: """Base class for validation gates.""" def __init__(self, memory_store, next_gate=None): self.memory = memory_store self.next_gate = next_gate self.name = "GenericGate" # Override in subclasses async def validate(self, input_data, task_description, worker_id, metadata): """ Validates the input data and optionally passes it to the next gate. Args: input_data: The data to validate. task_description: Description of the task. worker_id: ID of the worker processing the task. metadata: Additional metadata. Returns: A tuple containing: - A boolean indicating whether the validation passed. - A dictionary of checks performed. - A report dictionary containing the validation results. """ raise NotImplementedError async def _log_decision(self, worker_id, decision, reasoning, input_data, metadata): """Logs the validation decision with reasoning.""" log_entry = { "timestamp": datetime.utcnow().isoformat(), "gate": self.name, "worker_id": worker_id, "decision": decision, "reasoning": reasoning, "input_data": str(input_data)[:500], # Truncate for brevity "metadata": str(metadata)[:500] # Truncate for brevity } try: self.memory.log_audit(f"{self.name}_validation", worker_id, log_entry) except Exception as e: print(f"[WARNING] Audit logging failed for {self.name}: {e}") async def _chain_validation(self, input_data, task_description, worker_id, metadata, previous_checks): """Chains the validation to the next gate if it exists.""" if self.next_gate: return await self.next_gate.validate(input_data, task_description, worker_id, metadata, previous_checks) else: return True, previous_checks, {"message": "Validation chain complete"} class GateAlpha(ValidationGate): """Gate Alpha: Input Validity - Verify source data quality.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateAlpha" async def validate(self, input_data, task_description, worker_id, metadata, previous_checks=None): checks = {} # Check if input_data is not None and is of type string checks["input_data_present"] = input_data is not None and isinstance(input_data, str) # Check if task_description is not None and is of type string checks["task_description_present"] = task_description is not None and isinstance(task_description, str) passed = all(checks.values()) reasoning = "Input data and task description are valid." if passed else "Input data or task description is invalid." await self._log_decision(worker_id, passed, reasoning, input_data, metadata) if passed: return await self._chain_validation(input_data, task_description, worker_id, metadata, checks) else: report = {"gate": self.name, "passed": False, "checks": checks, "reason":reasoning} return False, checks, report class GateBeta(ValidationGate): """Gate Beta: Output Quality - Check accuracy and completeness.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateBeta" async def validate(self, output, task_description, worker_id, metadata, previous_checks=None): checks = {} # Check if the output is not empty after removing whitespace checks["output_not_empty"] = output is not None and isinstance(output, str) and output.strip() != "" # Perform a simple keyword check based on the task description keywords = task_description.lower().split() output_lower = output.lower() checks["relevant_keywords_present"] = any(keyword in output_lower for keyword in keywords) passed = all(checks.values()) reasoning = "Output is accurate and complete." if passed else "Output is inaccurate or incomplete." await self._log_decision(worker_id, passed, reasoning, output, metadata) if passed: return await self._chain_validation(output, task_description, worker_id, metadata, checks) else: report = {"gate": self.name, "passed": False, "checks": checks, "reason":reasoning} return False, checks, report class GateGamma(ValidationGate): """Gate Gamma: Insight Purity - Detect hallucinations.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateGamma" async def validate(self, insight, task_description, worker_id, metadata, previous_checks=None): checks = {} # Simulate hallucination detection (replace with actual logic) checks["no_hallucinations"] = "hallucination_score" in metadata and metadata["hallucination_score"] < 0.5 passed = all(checks.values()) reasoning = "No hallucinations detected." if passed else "Hallucinations detected." await self._log_decision(worker_id, passed, reasoning, insight, metadata) if passed: return await self._chain_validation(insight, task_description, worker_id, metadata, checks) else: report = {"gate": self.name, "passed": False, "checks": checks, "reason":reasoning} return False, checks, report class GateDelta(ValidationGate): """Gate Delta: Memory Integration - Validate storage operations.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateDelta" async def validate(self, data, task_description, worker_id, metadata, previous_checks=None): checks = {} try: # Simulate memory write operation self.memory.store_data(worker_id, data) retrieved_data = self.memory.retrieve_data(worker_id) checks["memory_write_successful"] = retrieved_data == data except Exception as e: checks["memory_write_successful"] = False print(f"[ERROR] Memory operation failed: {e}") passed = all(checks.values()) reasoning = "Memory integration successful." if passed else "Memory integration failed." await self._log_decision(worker_id, passed, reasoning, data, metadata) if passed: return await self._chain_validation(data, task_description, worker_id, metadata, checks) else: report = {"gate": self.name, "passed": False, "checks": checks, "reason":reasoning} return False, checks, report class GateEpsilon(ValidationGate): """Gate Epsilon: Strategy Alignment - Ensure revenue pathway fit.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateEpsilon" async def validate(self, data, task_description, worker_id, metadata, previous_checks=None): checks = {} # Simulate strategy alignment check based on task description and metadata checks["strategy_aligned"] = "revenue_potential" in metadata and metadata["revenue_potential"] > 0.7 passed = all(checks.values()) reasoning = "Strategy is aligned with revenue pathway." if passed else "Strategy is not aligned with revenue pathway." await self._log_decision(worker_id, passed, reasoning, data, metadata) if passed: return await self._chain_validation(data, task_description, worker_id, metadata, checks) else: report = {"gate": self.name, "passed": False, "checks": checks, "reason":reasoning} return False, checks, report class GateZeta(ValidationGate): """Gate Zeta: Budget Compliance - Resource monitoring.""" def __init__(self, memory_store, next_gate=None): super().__init__(memory_store, next_gate) self.name = "GateZeta" async def validate(self, data, task_description, worker_id, metadata, previous_checks=None): checks = {} # Simulate budget compliance check based on resource usage in metadata checks["budget_compliant"] = "resource_usage" in metadata and metadata["resource_usage"] < 0.9 passed = all(checks.values()) reasoning = "Budget is compliant." if passed else "Budget is not compliant." await self._log_decision(worker_id, passed, reasoning, data, metadata) report = {"gate": self.name, "passed": passed, "checks": checks, "reason":reasoning} return passed, checks, report # --- Orchestrator --- class SixGateValidator: def __init__(self, memory): self.memory = memory # Chain the gates together self.gate_alpha = GateAlpha(self.memory) self.gate_beta = GateBeta(self.memory, self.gate_alpha) self.gate_gamma = GateGamma(self.memory, self.gate_beta) self.gate_delta = GateDelta(self.memory, self.gate_gamma) self.gate_epsilon = GateEpsilon(self.memory, self.gate_delta) self.gate_zeta = GateZeta(self.memory, self.gate_epsilon) #zeta is now the first gate async def validate_worker_output(self, output, task_description, worker_id, metadata): """Run complete 6-gate validation suite.""" start_time = datetime.utcnow() passed, checks, report = await self.gate_zeta.validate(output, task_description, worker_id, metadata) #start at the zeta gate end_time = datetime.utcnow() overall_report = { "timestamp": start_time.isoformat(), "worker_id": worker_id, "passed": passed, "gates": { "alpha": checks.get("GateAlpha", {}), "beta": checks.get("GateBeta", {}), "gamma": checks.get("GateGamma", {}), "delta": checks.get("GateDelta", {}), "epsilon": checks.get("GateEpsilon", {}), "zeta": checks.get("GateZeta", {}) }, "duration": (end_time - start_time).total_seconds() } # Log to immutable audit trail try: self.memory.log_audit("six_gate_validation", worker_id, overall_report) except Exception as e: print(f"[WARNING] Audit logging failed: {e}") # Update worker trust (Patent P9) - placeholder implementation try: self.memory.update_worker_trust(worker_id, passed) except Exception as e: print(f"[WARNING] Trust update failed: {e}") return overall_report # --- Example Usage (replace with your actual implementation) --- if __name__ == '__main__': # Mock memory store class MockMemory: def __init__(self): self.data = {} self.audit_logs = [] def store_data(self, key, value): self.data[key] = value def retrieve_data(self, key): return self.data.get(key) def log_audit(self, system, worker_id, log_entry): self.audit_logs.append((system, worker_id, log_entry)) print(f"Audit Log: {system}, {worker_id}, {log_entry}") def update_worker_trust(self, worker_id, passed): print(f"Worker {worker_id} trust updated to {passed}") memory = MockMemory() validator = SixGateValidator(memory) async def main(): output = "This is a sample output." task_description = "Summarize the input text." worker_id = "worker123" metadata = {"hallucination_score": 0.2, "revenue_potential": 0.8, "resource_usage": 0.6} report = await validator.validate_worker_output(output, task_description, worker_id, metadata) print(f"Validation Report: {report}") asyncio.run(main())