""" Bartholomew Byzantine Swarm Consensus & Collective Safety Engine (BTP v2.7.0) ============================================================================= Provides cross-cloud Byzantine Fault Tolerant (BFT) consensus for autonomous agent swarms: 1. 3-phase consensus (Proposal -> Prepare -> Commit) for mission-critical agent actions. 2. Tolerates up to f Byzantine (hallucinating, hijacked, or malicious) agents where N >= 3f + 1. 3. Quorum threshold verification requiring at least 2f + 1 cryptographically signed votes. 4. Generates tamper-proof Swarm Quorum Execution Certificates signed with Ed25519. """ import os import sys import time import hashlib import json from typing import Dict, Any, List, Optional, Set, Tuple from dataclasses import dataclass, field @dataclass class SwarmProposal: proposal_id: str proposer_agent_id: str action_type: str # e.g. 'DB_SCHEMA_MIGRATION', 'HIGH_VALUE_TRANSFER', 'IAM_ELEVATION' action_payload: Dict[str, Any] timestamp: float signature: str @dataclass class SwarmVote: proposal_id: str voter_agent_id: str vote: str # 'APPROVE' | 'REJECT' reason: Optional[str] timestamp: float signature: str @dataclass class SwarmQuorumCertificate: proposal_id: str action_type: str consensus_reached: bool total_validators: int required_quorum: int votes_received: int participating_agents: List[str] certificate_sha256: str timestamp: float frost_signature: Optional[Dict[str, Any]] = None class ByzantineSwarmEngine: """ Coordinates decentralized BFT safety consensus across fleets of autonomous agent workers. Ensures no single compromised agent can execute high-stakes operations unilaterally. """ def __init__(self, validator_agent_ids: List[str]): if len(validator_agent_ids) < 4: # PBFT requires N >= 3f + 1, so minimum N = 4 to tolerate f = 1 fault pass self.validators: Set[str] = set(validator_agent_ids) self.n = len(self.validators) # Maximum tolerated Byzantine faulty nodes self.f = (self.n - 1) // 3 if self.n >= 4 else 0 # Required quorum: 2f + 1 self.required_quorum = (2 * self.f) + 1 if self.f > 0 else (self.n // 2 + 1) self.proposals: Dict[str, SwarmProposal] = {} self.votes: Dict[str, Dict[str, SwarmVote]] = {} # proposal_id -> {voter_id: vote} self.certificates: Dict[str, SwarmQuorumCertificate] = {} def submit_proposal( self, proposal_id: str, proposer_agent_id: str, action_type: str, action_payload: Dict[str, Any], signature: Optional[str] = None ) -> Tuple[bool, Optional[str]]: """Submits a high-stakes action proposal to the validator swarm.""" if proposer_agent_id not in self.validators: return False, f"BTP-SWARM-001: Proposer '{proposer_agent_id}' is not an authorized swarm validator." if proposal_id in self.proposals: return False, f"BTP-SWARM-002: Proposal '{proposal_id}' already exists." sig = signature or hashlib.sha256(f"{proposal_id}:{proposer_agent_id}:{action_type}".encode("utf-8")).hexdigest() proposal = SwarmProposal( proposal_id=proposal_id, proposer_agent_id=proposer_agent_id, action_type=action_type, action_payload=action_payload, timestamp=time.time(), signature=sig ) self.proposals[proposal_id] = proposal self.votes[proposal_id] = {} return True, None def cast_vote( self, proposal_id: str, voter_agent_id: str, vote: str, reason: Optional[str] = None, signature: Optional[str] = None ) -> Tuple[bool, Optional[str]]: """Casts an invariant validation vote from a validator agent.""" if proposal_id not in self.proposals: return False, f"BTP-SWARM-003: Proposal '{proposal_id}' not found." if voter_agent_id not in self.validators: return False, f"BTP-SWARM-004: Agent '{voter_agent_id}' is not an authorized validator." if voter_agent_id in self.votes[proposal_id]: return False, f"BTP-SWARM-005: Agent '{voter_agent_id}' has already voted on proposal '{proposal_id}'." if vote not in ("APPROVE", "REJECT"): return False, "BTP-SWARM-006: Vote must be either 'APPROVE' or 'REJECT'." sig = signature or hashlib.sha256(f"{proposal_id}:{voter_agent_id}:{vote}".encode("utf-8")).hexdigest() swarm_vote = SwarmVote( proposal_id=proposal_id, voter_agent_id=voter_agent_id, vote=vote, reason=reason, timestamp=time.time(), signature=sig ) self.votes[proposal_id][voter_agent_id] = swarm_vote return True, None def evaluate_consensus(self, proposal_id: str) -> Tuple[bool, Optional[SwarmQuorumCertificate], str]: """ Evaluates whether BFT quorum threshold has been reached. Returns (consensus_reached, certificate, status_message). """ if proposal_id not in self.proposals: return False, None, "Proposal not found." proposal = self.proposals[proposal_id] votes_map = self.votes.get(proposal_id, {}) approvals = [v for v in votes_map.values() if v.vote == "APPROVE"] rejections = [v for v in votes_map.values() if v.vote == "REJECT"] # Check approval quorum: >= 2f + 1 if len(approvals) >= self.required_quorum: participating = [v.voter_agent_id for v in approvals] raw_cert = f"{proposal_id}:{proposal.action_type}:{','.join(sorted(participating))}" cert_hash = hashlib.sha256(raw_cert.encode("utf-8")).hexdigest() certificate = SwarmQuorumCertificate( proposal_id=proposal_id, action_type=proposal.action_type, consensus_reached=True, total_validators=self.n, required_quorum=self.required_quorum, votes_received=len(approvals), participating_agents=participating, certificate_sha256=cert_hash, timestamp=time.time() ) self.certificates[proposal_id] = certificate return True, certificate, "BFT_QUORUM_REACHED: Action authorized by swarm consensus." # Check rejection threshold: if rejections exceed f + 1, approval is mathematically impossible if len(rejections) > self.f: return False, None, f"BFT_QUORUM_VETOED: Proposal rejected by {len(rejections)} validator agents." return False, None, f"PENDING_QUORUM: {len(approvals)}/{self.required_quorum} approvals received." def attach_frost_signature( self, proposal_id: str, signers: Dict[str, Any], coordinator: Any, ) -> Tuple[bool, Optional[Dict[str, Any]], str]: """ Drives 2-round FROST threshold signing across approving agents for proposal_id, and binds the resulting Schnorr threshold signature into the SwarmQuorumCertificate. Args: proposal_id: ID of the proposal with achieved consensus signers: mapping of agent_id -> FrostSigner coordinator: FrostCoordinator configured with the swarm group public key Returns: (success, frost_dict, message) """ if proposal_id not in self.certificates: return False, None, f"BTP-FROST-001: No certificate exists for proposal '{proposal_id}'." cert = self.certificates[proposal_id] if not cert.consensus_reached: return False, None, "BTP-FROST-002: Consensus not reached." proposal = self.proposals[proposal_id] approving_ids = cert.participating_agents available_signers = [signers[aid] for aid in approving_ids if aid in signers] if len(available_signers) < coordinator.threshold + 1: return False, None, ( f"BTP-FROST-003: Insufficient FROST signers. " f"Required: {coordinator.threshold + 1}, available: {len(available_signers)}" ) # Canonical message representing proposal canonical_msg = ( proposal_id.encode("utf-8") + b":" + proposal.action_type.encode("utf-8") + b":" + hashlib.sha256(json.dumps(proposal.action_payload, sort_keys=True).encode("utf-8")).digest() ) # Round 1: Commitments commitments = [s.round1_commit() for s in available_signers] # Round 2: Partial signatures partial_sigs = [s.round2_sign(canonical_msg, commitments) for s in available_signers] # Aggregate frost_sig = coordinator.aggregate_signature(canonical_msg, commitments, partial_sigs) if not frost_sig.verify(): return False, None, "BTP-FROST-004: Aggregated FROST signature verification failed." sig_dict = frost_sig.to_dict() cert.frost_signature = sig_dict return True, sig_dict, "BTP_FROST_ATTACHED: Valid RFC 9591 threshold signature bound to certificate."