import datetime import logging # Configure logging for better traceability in AIVA's core logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class MultiModelConsensusSystem: """ Manages a 3-way multi-model voting system for high-stakes decisions within AIVA's Genesis Prime Mother evolution. Ensures Patent P5 compliance for critical validations. """ def __init__(self, required_models=['model_alpha', 'model_beta', 'model_gamma']): """ Initializes the consensus system. :param required_models: A list of model identifiers that must vote. """ if len(required_models) != 3: raise ValueError("Multi-Model Consensus requires exactly 3 models for high-stakes decisions.") self.required_models = set(required_models) self.voting_records = {} # Stores votes for each decision_id logging.info(f"MultiModelConsensusSystem initialized with models: {', '.join(self.required_models)}") def register_decision(self, decision_id: str, decision_details: dict): """ Registers a new high-stakes decision, preparing it for voting. :param decision_id: A unique identifier for the decision. :param decision_details: A dictionary containing details about the decision. :return: True if registered successfully, False if decision_id already exists. """ if decision_id in self.voting_records: logging.warning(f"Decision '{decision_id}' already registered. Skipping registration.") return False, "Decision already registered." self.voting_records[decision_id] = { "details": decision_details, "votes": {model_id: None for model_id in self.required_models}, "timestamp_initiated": datetime.datetime.now().isoformat(), "status": "PENDING", # PENDING, IN_CONSENSUS, DISPUTED, RESOLVED "consensus_result": None, "resolution_log": [] } logging.info(f"Decision '{decision_id}' registered for voting with details: {decision_details.get('description', 'N/A')}") return True, "Decision registered." def submit_vote(self, decision_id: str, model_id: str, vote_value: bool): """ Submits a vote for a specific decision by a specific model. :param decision_id: Unique identifier for the decision. :param model_id: Identifier of the model casting the vote. :param vote_value: The vote (True for approve, False for reject). :return: A tuple (success: bool, message: str). """ if model_id not in self.required_models: logging.warning(f"Unauthorized model '{model_id}' attempted to vote on decision '{decision_id}'.") return False, "Unauthorized model." if decision_id not in self.voting_records: logging.error(f"Decision '{decision_id}' not registered. Cannot submit vote.") return False, "Decision not registered." record = self.voting_records[decision_id] if record["status"] in ["IN_CONSENSUS", "RESOLVED"]: logging.warning(f"Vote submitted for decision '{decision_id}' which is already {record['status']}. Vote ignored.") return False, f"Decision already {record['status']}." if record["votes"][model_id] is not None: logging.warning(f"Model '{model_id}' already voted for decision '{decision_id}'. Overwriting.") record["votes"][model_id] = { "value": vote_value, "timestamp": datetime.datetime.now().isoformat() } logging.info(f"Model '{model_id}' voted '{vote_value}' for decision '{decision_id}'.") # After each vote, check for consensus self._evaluate_consensus(decision_id) return True, "Vote recorded." def _evaluate_consensus(self, decision_id: str): """Internal method to check for consensus after a vote or status change.""" record = self.voting_records.get(decision_id) if not record: return # Filter out None votes to count only actual votes actual_votes = [v["value"] for m, v in record["votes"].items() if v is not None] models_voted_count = len(actual_votes) if models_voted_count == len(self.required_models): # All models have voted if all(vote == True for vote in actual_votes): record["status"] = "IN_CONSENSUS" record["consensus_result"] = True logging.info(f"Decision '{decision_id}' achieved full 'True' consensus.") elif all(vote == False for vote in actual_votes): record["status"] = "IN_CONSENSUS" record["consensus_result"] = False logging.info(f"Decision '{decision_id}' achieved full 'False' consensus.") else: record["status"] = "DISPUTED" record["consensus_result"] = None # Indicates no unanimous consensus logging.warning(f"Decision '{decision_id}' is DISPUTED: {record['votes']}. Initiating dispute resolution protocol.") else: record["status"] = "PENDING" logging.debug(f"Decision '{decision_id}' still PENDING. Votes received: {models_voted_count}/{len(self.required_models)}") def get_decision_status(self, decision_id: str): """ Retrieves the current status and voting record for a decision. :param decision_id: Unique identifier for the decision. :return: A dictionary with decision details, votes, and status, or None if not found. """ record = self.voting_records.get(decision_id) if record: # Return a copy to prevent external modification of internal state return { "decision_id": decision_id, "details": record["details"], "votes": {model: (vote["value"] if vote else None) for model, vote in record["votes"].items()}, "raw_votes": record["votes"], # For detailed timestamp info "status": record["status"], "consensus_result": record.get("consensus_result"), "timestamp_initiated": record["timestamp_initiated"], "resolution_log": record["resolution_log"] } return None def resolve_dispute(self, decision_id: str, resolution_action: str, resolved_value: bool = None, resolver_id: str = "AIVA_CORE_OVERRIDE"): """ Implements the dispute resolution protocol for a high-stakes decision. This protocol can involve human intervention, re-evaluation, or AIVA's self-correction. :param decision_id: Unique identifier for the disputed decision. :param resolution_action: Description of the action taken (e.g., "Human Override", "Re-evaluation Triggered"). :param resolved_value: The final determined value if the dispute is resolved (True/False/None). If None, the dispute remains open but resolution action is logged. :param resolver_id: Identifier of the entity resolving the dispute. :return: A tuple (success: bool, message: str). """ record = self.voting_records.get(decision_id) if not record: logging.error(f"Cannot resolve dispute for unknown decision '{decision_id}'.") return False, "Decision not found." if record["status"] not in ["DISPUTED", "PENDING"]: logging.warning(f"Decision '{decision_id}' is not in DISPUTED or PENDING status (current: {record['status']}). Resolution logged but status not changed to RESOLVED automatically unless resolved_value is provided.") log_entry = { "action": resolution_action, "resolved_value": resolved_value, "resolver_id": resolver_id, "timestamp": datetime.datetime.now().isoformat(), "previous_status": record["status"] } record["resolution_log"].append(log_entry) if resolved_value is not None: record["status"] = "RESOLVED" record["consensus_result"] = resolved_value logging.info(f"Dispute for decision '{decision_id}' resolved by '{resolver_id}' with value '{resolved_value}'. Status: RESOLVED.") else: # If resolved_value is None, it means the action might be to re-evaluate, not a direct resolution # Status remains DISPUTED or PENDING, or is changed by the action itself if it triggers new votes. logging.info(f"Dispute for decision '{decision_id}' received resolution action '{resolution_action}' from '{resolver_id}'. Value not explicitly set, status remains '{record['status']}'.") return True, "Dispute resolution logged." def get_voting_record_full(self): """ Retrieves the complete internal voting records for audit purposes. For AIVA's evolution, this provides transparency and learning data. """ return self.voting_records