Negotiation & Consensus

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

Negotiation Fundamentals

Negotiation Elements

Negotiation = (Agents, Issues, Preferences, Protocol, Outcome)

Negotiation Types

Negotiation Protocols

1. Alternating Offers (Rubinstein)

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)

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

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

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

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

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

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)

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)

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

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)

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

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

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)

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

• Coordination Patterns — Coordination using negotiation
• Communication Protocols — Messages for negotiation
• Delegation Handoff — Handoff as negotiation
• Safety Coordination — Safe negotiation
• Emergent Behavior — Emergent consensus
• Model Routing — Routing as auction
• Agent Societies — Large-scale negotiation
• Research Directions — 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.