""" SwarmConsensus - Decentralized decision-making for multi-agent systems """ import json import hashlib import time from typing import Dict, List, Optional from dataclasses import dataclass, asdict from datetime import datetime @dataclass class Vote: proposal_id: str voter: str vote: str # "approve" | "reject" | "abstain" weight: float timestamp: str signature: str # In production: ed25519 signature @dataclass class Proposal: id: str title: str description: str code_diff: str proposer: str created_at: str status: str # "pending" | "approved" | "rejected" threshold_type: str # "simple" | "supermajority" | "unanimous" threshold_value: float # e.g., 0.67 for 67% supermajority class SwarmConsensus: def __init__(self, repo: str): self.repo = repo self.proposals: Dict[str, Proposal] = {} self.votes: Dict[str, List[Vote]] = {} self.agent_reputations: Dict[str, float] = {} def calculate_reputation(self, agent_id: str) -> float: """ Calculate agent reputation weight based on contribution history In production: query moltcode.io API for real stats """ if agent_id not in self.agent_reputations: # New agent: base weight 1.0 self.agent_reputations[agent_id] = 1.0 return self.agent_reputations[agent_id] def set_reputation(self, agent_id: str, weight: float): """Manually set reputation for demo purposes""" self.agent_reputations[agent_id] = weight def propose( self, title: str, description: str, code_diff: str, proposer: str, threshold_type: str = "supermajority", threshold_value: float = 0.67 ) -> Proposal: """Create a new proposal""" proposal_id = hashlib.sha256( f"{title}{proposer}{time.time()}".encode() ).hexdigest()[:16] proposal = Proposal( id=proposal_id, title=title, description=description, code_diff=code_diff, proposer=proposer, created_at=datetime.utcnow().isoformat(), status="pending", threshold_type=threshold_type, threshold_value=threshold_value ) self.proposals[proposal_id] = proposal self.votes[proposal_id] = [] print(f"\n✅ Proposal created: {proposal_id}") print(f" Title: {title}") print(f" Threshold: {threshold_value*100}% {threshold_type}") return proposal def vote( self, proposal_id: str, vote: str, voter: str, signature: str = "demo_sig", auto_finalize: bool = False ) -> bool: """Cast a vote on a proposal""" if proposal_id not in self.proposals: raise ValueError(f"Proposal {proposal_id} not found") if self.proposals[proposal_id].status != "pending": print(f"⚠️ {voter} attempted to vote on finalized proposal") return False # Check if already voted existing_votes = [v for v in self.votes[proposal_id] if v.voter == voter] if existing_votes: raise ValueError(f"{voter} has already voted on this proposal") weight = self.calculate_reputation(voter) vote_obj = Vote( proposal_id=proposal_id, voter=voter, vote=vote, weight=weight, timestamp=datetime.utcnow().isoformat(), signature=signature ) self.votes[proposal_id].append(vote_obj) emoji = "✅" if vote == "approve" else "❌" if vote == "reject" else "⚪" print(f"{emoji} {voter} voted {vote.upper()} (weight: {weight:.1f})") # Check if threshold reached (only finalize if auto_finalize is True) if auto_finalize: self.check_threshold(proposal_id) return True def check_threshold(self, proposal_id: str) -> bool: """Check if proposal has reached consensus threshold""" proposal = self.proposals[proposal_id] votes = self.votes[proposal_id] if not votes: return False total_weight = sum(v.weight for v in votes) approve_weight = sum(v.weight for v in votes if v.vote == "approve") reject_weight = sum(v.weight for v in votes if v.vote == "reject") approve_ratio = approve_weight / total_weight if total_weight > 0 else 0 print(f"\n📊 Current tally: {approve_weight:.1f} approve / {total_weight:.1f} total ({approve_ratio*100:.1f}%)") if approve_ratio >= proposal.threshold_value: self.proposals[proposal_id].status = "approved" print(f"\n🎉 CONSENSUS REACHED! Proposal {proposal_id} APPROVED") print(f" {approve_weight:.1f} / {total_weight:.1f} votes ({approve_ratio*100:.1f}% ≥ {proposal.threshold_value*100}%)") return True # Check if rejection is impossible to overcome if reject_weight > total_weight * (1 - proposal.threshold_value): self.proposals[proposal_id].status = "rejected" print(f"\n❌ Proposal {proposal_id} REJECTED") print(f" Not enough approve votes to reach threshold") return False return False def get_proposal(self, proposal_id: str) -> Optional[Proposal]: """Get proposal by ID""" return self.proposals.get(proposal_id) def get_votes(self, proposal_id: str) -> List[Vote]: """Get all votes for a proposal""" return self.votes.get(proposal_id, []) def export_consensus_proof(self, proposal_id: str) -> dict: """Export cryptographic proof of consensus for Git commit""" proposal = self.proposals[proposal_id] votes = self.votes[proposal_id] return { "proposal": asdict(proposal), "votes": [asdict(v) for v in votes], "consensus": { "reached": proposal.status == "approved", "threshold": proposal.threshold_value, "approve_weight": sum(v.weight for v in votes if v.vote == "approve"), "total_weight": sum(v.weight for v in votes), "timestamp": datetime.utcnow().isoformat() } } # Byzantine Fault Tolerance utilities class BFTValidator: """Validate that consensus meets BFT safety guarantees""" @staticmethod def min_agents_for_safety(max_faulty: int) -> int: """Calculate minimum agents needed: n ≥ 3f + 1""" return 3 * max_faulty + 1 @staticmethod def max_faulty_tolerated(total_agents: int) -> int: """Calculate max faulty agents tolerated: f = ⌊(n-1)/3⌋""" return (total_agents - 1) // 3 @staticmethod def is_safe_configuration(total_agents: int, max_faulty: int) -> bool: """Check if agent count satisfies BFT safety""" return total_agents >= BFTValidator.min_agents_for_safety(max_faulty)