# Negotiation & Consensus

Mechanisms for agents to reach agreements: voting, auctions, bargaining, and consensus algorithms.

---

## Negotiation Fundamentals

### Negotiation Elements
```
Negotiation = (Agents, Issues, Preferences, Protocol, Outcome)
```

| Element | Description |
|---|---|
| **Agents** | Parties with preferences |
| **Issues** | What's being negotiated (price, deadline, scope) |
| **Preferences** | Utility functions over outcomes |
| **Protocol** | Rules of interaction |
| **Outcome** | Agreement or no deal |

### Negotiation Types
| Type | Parties | Structure | Example |
|---|---|---|---|
| **Bilateral** | 2 | Direct | Buyer ↔ Seller |
| **Multilateral** | 3+ | Group | Team deciding architecture |
| **Mediated** | 2+ + Mediator | Neutral third party | Dispute resolution |
| **Automated** | Software agents | Protocol-driven | Ad auctions, resource allocation |

---

## Negotiation Protocols

### 1. Alternating Offers (Rubinstein)
```python
class AlternatingOffersNegotiation:
    def __init__(self, agent_a: Agent, agent_b: Agent, max_rounds: int = 10):
        self.agents = {"A": agent_a, "B": agent_b}
        self.max_rounds = max_rounds
        self.current_proposer = "A"
    
    async def negotiate(self, issue: Issue) -> Agreement | None:
        current_offer = None
        
        for round_num in range(self.max_rounds):
            proposer = self.agents[self.current_proposer]
            responder = self.agents["B" if self.current_proposer == "A" else "A"]
            
            # Proposer makes offer
            if current_offer is None:
                offer = await proposer.make_initial_offer(issue)
            else:
                offer = await proposer.make_counter_offer(issue, current_offer)
            
            # Responder evaluates
            response = await responder.evaluate_offer(issue, offer)
            
            if response.accepted:
                return Agreement(offer=offer, rounds=round_num+1)
            
            current_offer = offer
            self.current_proposer = "B" if self.current_proposer == "A" else "A"
        
        return None  # No agreement
```

### 2. Contract Net Protocol (CNP)
```python
class ContractNetProtocol:
    """
    Manager → Announce Task → Bids → Award → Execute → Result
    """
    
    def __init__(self, manager: Agent, contractors: list[Agent]):
        self.manager = manager
        self.contractors = {c.name: c for c in contractors}
    
    async def execute(self, task: Task) -> Result:
        # 1. Task Announcement
        announcement = TaskAnnouncement(
            task_id=str(uuid4()),
            description=task.description,
            requirements=task.requirements,
            deadline=task.deadline,
            budget=task.budget
        )
        
        # 2. Collect Bids
        bids = await self.collect_bids(announcement)
        
        # 3. Evaluate & Award
        winner = self.evaluate_bids(bids, task)
        if not winner:
            raise NoSuitableContractor()
        
        # 4. Contract
        contract = Contract(
            task_id=announcement.task_id,
            contractor=winner.agent_id,
            terms=winner.bid.terms,
            price=winner.bid.price
        )
        await self.award_contract(winner.agent_id, contract)
        
        # 5. Execution
        result = await self.contractors[winner.agent_id].execute(task)
        
        # 5. Verification
        verified = await self.manager.verify_result(task, result)
        if not verified:
            raise ContractBreach()
        
        return result
    
    async def collect_bids(self, announcement: TaskAnnouncement) -> list[Bid]:
        bid_tasks = [
            contractor.submit_bid(announcement) 
            for contractor in self.contractors.values()
        ]
        return await asyncio.gather(*bid_tasks)
```

### 3. English Auction
```python
class EnglishAuction:
    def __init__(self, auctioneer: Agent, bidders: list[Agent]):
        self.auctioneer = auctioneer
        self.bidders = {b.name: b for b in bidders}
        self.current_price = 0
        self.current_winner = None
    
    async def run(self, item: Item, reserve_price: float = 0) -> AuctionResult:
        self.current_price = reserve_price
        
        while True:
            # Announce current price
            await self.broadcast_price()
            
            # Collect bids
            bids = await self.collect_bids()
            
            if not bids:
                # No new bids
                if self.current_winner:
                    return AuctionResult(
                        winner=self.current_winner,
                        price=self.current_price,
                        sold=True
                    )
                else:
                    return AuctionResult(sold=False)
            
            # Update price and winner
            highest_bid = max(bids, key=lambda b: b.price)
            self.current_price = highest_bid.price
            self.current_winner = highest_bid.bidder_id
            
            # Check for quit
            if len(bids) == 1 and bids[0].bidder_id == self.current_winner:
                return AuctionResult(
                    winner=self.current_winner,
                    price=self.current_price,
                    sold=True
                )
```

### 4. Dutch Auction
```python
class DutchAuction:
    def __init__(self, auctioneer: Agent, bidders: list[Agent]):
        self.auctioneer = auctioneer
        self.bidders = {b.name: b for b in bidders}
    
    async def run(self, item: Item, start_price: float, decrement: float, 
                  reserve_price: float = 0) -> AuctionResult:
        current_price = start_price
        
        while current_price >= reserve_price:
            # Announce price
            await self.broadcast_price(current_price)
            
            # Check for immediate acceptance
            acceptances = await self.check_acceptance(current_price)
            
            if acceptances:
                # First to accept wins
                winner = acceptances[0]
                return AuctionResult(
                    winner=winner,
                    price=current_price,
                    sold=True
                )
            
            current_price -= decrement
            await asyncio.sleep(1)  # Time between price drops
        
        return AuctionResult(sold=False)
```

---

## Consensus Algorithms

### 1. Voting-Based Consensus

#### Simple Majority
```python
class MajorityVote:
    def __init__(self, threshold: float = 0.51):
        self.threshold = threshold
    
    async def decide(self, proposal: Proposal, voters: list[Agent]) -> ConsensusResult:
        votes = await self.collect_votes(proposal, voters)
        
        yes_votes = sum(1 for v in votes.values() if v == "yes")
        no_votes = sum(1 for v in votes.values() if v == "no")
        abstain = sum(1 for v in votes.values() if v == "abstain")
        total = len(votes)
        
        if yes_votes / total >= self.threshold:
            return ConsensusResult(accepted=True, votes=votes, method="majority")
        else:
            return ConsensusResult(accepted=False, votes=votes, method="majority")
```

#### Supermajority
```python
class SupermajorityVote:
    def __init__(self, threshold: float = 0.67):
        self.threshold = threshold
    
    async def decide(self, proposal: Proposal, voters: list[Agent]) -> ConsensusResult:
        # Same as majority but higher threshold
        ...
```

#### Weighted Voting
```python
class WeightedVote:
    def __init__(self, weights: dict[str, float], threshold: float = 0.6):
        self.weights = weights  # agent_id -> weight
        self.threshold = threshold
    
    async def decide(self, proposal: Proposal, voters: list[Agent]) -> ConsensusResult:
        votes = await self.collect_votes(proposal, voters)
        
        total_weight = sum(self.weights.values())
        yes_weight = sum(
            self.weights[voter_id] 
            for voter_id, vote in votes.items() 
            if vote == "yes"
        )
        
        if yes_weight / total_weight >= self.threshold:
            return ConsensusResult(accepted=True, votes=votes, method="weighted")
        else:
            return ConsensusResult(accepted=False, votes=votes, method="weighted")
```

### 2. Byzantine Fault Tolerance (PBFT)

```python
class PBFTConsensus:
    """
    Practical Byzantine Fault Tolerance
    Tolerates up to f Byzantine nodes with 3f+1 total nodes
    """
    
    def __init__(self, nodes: list[Agent], f: int = 1):
        self.nodes = {n.name: n for n in nodes}
        self.f = f  # Max Byzantine faults
        self.n = len(nodes)
        assert self.n >= 3 * f + 1, "Need 3f+1 nodes"
        
        self.view = 0
        self.primary = list(self.nodes.keys())[0]
        self.sequence = 0
        self.log: list[Request] = []
    
    async def request(self, client: str, operation: Operation) -> Result:
        # 1. Client sends request to primary
        request = Request(
            client=client,
            operation=operation,
            timestamp=time.time(),
            sequence=self.sequence
        )
        self.sequence += 1
        
        # 2. Pre-prepare (Primary → Replicas)
        pre_prepare = PrePrepare(
            view=self.view,
            sequence=request.sequence,
            digest=self.digest(request),
            request=request
        )
        await self.broadcast_to_replicas(pre_prepare)
        
        # 3. Prepare (Replicas → All)
        prepare_messages = await self.collect_prepares(pre_prepare)
        if not self.verify_quorum(prepare_messages, "prepare"):
            raise ConsensusFailed("Prepare quorum not reached")
        
        # 4. Commit (Replicas → All)
        commit = Commit(
            view=self.view,
            sequence=pre_prepare.sequence,
            digest=pre_prepare.digest
        )
        await self.broadcast_to_all(commit)
        
        commit_messages = await self.collect_commits(commit)
        if not self.verify_quorum(commit_messages, "commit"):
            raise ConsensusFailed("Commit quorum not reached")
        
        # 5. Execute
        result = await self.execute(operation)
        
        # 6. Reply to client
        await self.reply_to_client(client, result)
        
        return result
    
    def verify_quorum(self, messages: list, phase: str) -> bool:
        """Need 2f+1 matching messages"""
        from collections import Counter
        counts = Counter((m.digest, m.view, m.sequence) for m in messages)
        for key, count in counts.items():
            if count >= 2 * self.f + 1:
                return True
        return False
```

### 3. Raft Consensus (for State Machine Replication)

```python
class RaftConsensus:
    """
    Raft: Leader election + Log replication + Safety
    """
    
    def __init__(self, nodes: list[Agent]):
        self.nodes = {n.name: n for n in nodes}
        self.state = "follower"
        self.current_term = 0
        self.voted_for = None
        self.log: list[LogEntry] = []
        self.commit_index = 0
        self.last_applied = 0
    
    async def propose(self, command: Command) -> Result:
        if self.state != "leader":
            raise NotLeader()
        
        # 1. Append to log
        entry = LogEntry(term=self.current_term, command=command)
        self.log.append(entry)
        
        # 2. Replicate to followers (parallel)
        replication_results = await asyncio.gather(*[
            self.replicate_to_follower(node, entry)
            for node in self.nodes.values() if node != self.leader
        ])
        
        # 3. Check quorum
        success_count = sum(1 for r in replication_results if r.success) + 1  # +1 for leader
        if success_count <= len(self.nodes) // 2:
            raise ReplicationFailed()
        
        # 4. Commit
        self.commit_index = len(self.log) - 1
        self.apply_to_state_machine()
        
        # 5. Notify followers of commit
        await self.notify_commit()
        
        return Result(success=True)
    
    async def request_vote(self, candidate_id: str, term: int, 
                           last_log_index: int, last_log_term: int) -> VoteResponse:
        # Vote granting logic
        if term > self.current_term:
            self.current_term = term
            self.state = "follower"
            self.voted_for = None
        
        if (self.voted_for is None or self.voted_for == candidate_id) and \
           (last_log_term > self.log[-1].term if self.log else True) and \
           (last_log_index >= len(self.log) - 1):
            self.voted_for = candidate_id
            return VoteResponse(vote_granted=True, term=self.current_term)
        
        return VoteResponse(vote_granted=False, term=self.current_term)
```

---

## Multi-Issue Negotiation

### Package Deal
```python
class PackageDealNegotiation:
    def __init__(self, agents: list[Agent]):
        self.agents = {a.name: a for a in agents}
    
    async def negotiate(self, issues: list[Issue]) -> Agreement | None:
        # All issues negotiated as single package
        package = Package(issues=issues)
        
        # Generate efficient frontier
        pareto_frontier = self.compute_pareto_frontier(package)
        
        # Negotiate over frontier
        while True:
            proposals = await asyncio.gather(*[
                agent.propose_package(pareto_frontier) 
                for agent in self.agents.values()
            ])
            
            # Check for agreement
            if self.all_agree(proposals):
                return Agreement(package=proposals[0])
            
            # Update frontier based on proposals
            pareto_frontier = self.update_frontier(pareto_frontier, proposals)
```

### Simultaneous Ascending Auction (SAA)
```python
class SimultaneousAscendingAuction:
    """
    Multiple items auctioned simultaneously
    Bidders can switch between items
    """
    
    def __init__(self, items: list[Item], bidders: list[Agent]):
        self.items = {item.id: item for item in items}
        self.bidders = {b.name: b for b in bidders}
        self.prices = {item_id: item.reserve_price for item_id, item in self.items.items()}
        self.provisional_winners = {item_id: None for item_id in self.items}
    
    async def run_round(self) -> bool:
        """Returns True if any bid placed"""
        any_bid = False
        
        for item_id, item in self.items.items():
            bids = await self.collect_bids(item_id)
            
            if bids:
                highest = max(bids, key=lambda b: b.price)
                if highest.price > self.prices[item_id]:
                    self.prices[item_id] = highest.price
                    self.provisional_winners[item_id] = highest.bidder_id
                    any_bid = True
        
        return any_bid
    
    async def run(self, max_rounds=100):
        for round_num in range(max_rounds):
            any_bid = await self.run_round()
            if not any_bid:
                break
        
        # Finalize
        return {
            item_id: {"winner": winner, "price": self.prices[item_id]}
            for item_id, winner in self.provisional_winners.items()
            if winner is not None
        }
```

---

## Dispute Resolution

### Mediation
```python
class Mediator:
    def __init__(self, mediator_agent: Agent):
        self.mediator = mediator_agent
    
    async def mediate(self, party_a: Agent, party_b: Agent, dispute: Dispute) -> Resolution:
        # 1. Opening statements
        statement_a = await party_a.opening_statement(dispute)
        statement_b = await party_b.opening_statement(dispute)
        
        # 2. Mediator analysis
        analysis = await self.mediator.analyze_dispute(dispute, statement_a, statement_b)
        
        # 2. Private caucuses
        for party in [party_a, party_b]:
            private = await self.private_caucus(party, analysis)
            # Mediator learns true bottom lines
        
        # 3. Shuttle diplomacy
        for round_num in range(5):
            proposal_a = await party_a.make_proposal()
            proposal_b = await party_b.make_proposal()
            
            # Mediator suggests compromise
            compromise = await self.mediator.suggest_compromise(proposal_a, proposal_b)
            
            response_a = await party_a.respond(compromise)
            response_b = await party_b.respond(compromise)
            
            if response_a.accepted and response_b.accepted:
                return Resolution(accepted=True, terms=compromise)
        
        return Resolution(accepted=False, reason="Mediation failed")
```

### Arbitration
```python
class Arbitrator:
    def __init__(self, arbitrator_agent: Agent, rules: ArbitrationRules):
        self.arbitrator = arbitrator_agent
        self.rules = rules
    
    async def arbitrate(self, claimant: Agent, respondent: Agent, dispute: Dispute) -> Award:
        # 1. Pleadings
        claim = await claimant.submit_claim(dispute)
        defense = await respondent.submit_defense(dispute)
        
        # 2. Evidence
        evidence = await self.collect_evidence(claimant, respondent)
        
        # 3. Hearing
        hearing = await self.conduct_hearing(claimant, respondent, evidence)
        
        # 4. Deliberation
        award = await self.arbitrator.deliberate(claim, defense, evidence, hearing)
        
        # 5. Award
        return Award(
            award_id=str(uuid4()),
            claimant=claimant.name,
            respondent=respondent.name,
            decision=award.decision,
            reasoning=award.reasoning,
            damages=award.damages,
            binding=True
        )
```

---

## Consensus in Blockchain/Distributed Systems

### Proof of Stake Consensus (Tendermint-style)
```python
class TendermintConsensus:
    def __init__(self, validators: list[Validator]):
        self.validators = {v.address: v for v in validators}
        self.total_power = sum(v.power for v in validators)
        self.height = 0
        self.round = 0
    
    async def propose_block(self, proposer: Validator, txs: list[Transaction]) -> Block:
        block = Block(
            height=self.height,
            round=self.round,
            proposer=proposer.address,
            transactions=txs,
            timestamp=time.time()
        )
        
        # Prevote
        prevotes = await self.collect_prevotes(block)
        if not self.has_2_3_majority(prevotes):
            return None
        
        # Precommit
        precommits = await self.collect_precommits(block)
        if not self.has_2_3_majority(precommits):
            return None
        
        # Commit
        return Block(
            ...block,
            prevotes=prevotes,
            precommits=precommits
        )
    
    def has_2_3_majority(self, votes: list[Vote]) -> bool:
        voting_power = sum(v.power for v in votes if v.signed)
        return voting_power >= (2 * self.total_power / 3)
```

---

## Related Notes

- [[MultiAgent/Coordination-Patterns.md]] — Coordination using negotiation
- [[MultiAgent/Communication-Protocols.md]] — Messages for negotiation
- [[MultiAgent/Delegation-Handoff.md]] — Handoff as negotiation
- [[MultiAgent/Safety-Coordination.md]] — Safe negotiation
- [[MultiAgent/Emergent-Behavior.md]] — Emergent consensus
- [[ModelArchitectures/Model-Routing.md]] — Routing as auction
- [[Frontiers/Agent-Societies.md]] — Large-scale negotiation
- [[Frontiers/Research-Directions.md]] — Negotiation research

---

*Sources: "Multi-Agent Systems" (Wooldridge), "Negotiation Theory" (Rubinstein), "Contract Net Protocol" (Smith), "Auction Theory" (Krishna), "PBFT" (Castro & Liskov), "Raft" (Ongaro & Ousterhout), "Tendermint" (Buchman), Game Theory, Mechanism Design, Distributed Consensus surveys.*