API Enhancement: Multi-Robot Consensus Protocol for FABRIC

[lau90eth]

# API Enhancement Proposal: Multi-Robot Consensus Protocol for FABRIC

## Summary
Distributed consensus protocol API that enables multiple OM1 agents to make coordinated decisions in real-time when operating in shared environments through FABRIC.

## Problem Statement

OM1 + FABRIC currently enables robots to verify identity, share context, and exchange skills. However, **there's no standardized mechanism for distributed decision-making when multiple robots encounter conflicting goals or need to coordinate on shared resources.**

### Real-World Scenarios Requiring Consensus:
1. **Resource Conflict**: Two warehouse robots identify the same object to pick up
2. **Path Planning**: Multiple robots navigating through a narrow corridor simultaneously  
3. **Task Assignment**: Fleet of delivery robots dynamically distributing new orders
4. **Collaborative Actions**: Humanoid robots coordinating to carry a large object together
5. **Safety Critical**: Multiple robots detecting different hazards need to agree on priority response

Without consensus, robots either:
- Operate independently causing conflicts/collisions
- Require centralized coordination (single point of failure)
- Use simple priority systems (inefficient resource utilization)

## Proposed Solution

### Core API Design

```python
# src/fabric/consensus.py

from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Callable

class ConsensusAlgorithm(Enum):
    """Supported consensus algorithms"""
    RAFT = "raft"              # Leader-based, crash fault tolerant
    PBFT = "pbft"              # Byzantine fault tolerant
    GOSSIP = "gossip"          # Scalable to 100+ robots
    WEIGHTED_VOTE = "weighted" # Priority-based (battery, capability, urgency)

@dataclass
class Proposal:
    """A proposal for robots to vote on"""
    proposal_id: str
    proposer_agent_id: str
    action: str              # e.g., "pick_object", "take_path_A"
    resource_id: str         # e.g., object_uuid, corridor_id
    priority: float          # 0.0-1.0
    timestamp: float
    metadata: Dict

class ConsensusProtocol:
    """Main consensus protocol interface for FABRIC"""
    
    def __init__(
        self, 
        agent_id: str,
        algorithm: ConsensusAlgorithm = ConsensusAlgorithm.RAFT,
        timeout: float = 5.0,
        quorum_size: Optional[int] = None
    ):
        self.agent_id = agent_id
        self.algorithm = algorithm
        self.timeout = timeout
        
    async def propose(self, proposal: Proposal) -> bool:
        """
        Propose action to robot network.
        Returns: True if consensus reached and approved
        """
        await self._broadcast_to_fabric(proposal)
        return await self._collect_votes(proposal.proposal_id)
    
    async def vote(self, proposal: Proposal, vote: bool, reason: Optional[str] = None):
        """Vote on another agent's proposal"""
        await self._send_vote_to_fabric(proposal.proposal_id, vote, reason)
    
    def register_vote_handler(self, handler: Callable[[Proposal], bool]):
        """Register custom voting logic"""
        self.vote_handler = handler

Integration Example

# Example: Pick object with consensus
from fabric.consensus import ConsensusProtocol, Proposal

class PickObjectAction:
    def __init__(self, agent_config):
        self.consensus = ConsensusProtocol(
            agent_id=agent_config["agent_id"],
            algorithm=ConsensusAlgorithm.WEIGHTED_VOTE
        )
        
        # Custom voting: prefer closer robots
        self.consensus.register_vote_handler(
            lambda p: p.priority > self.get_distance_to(p.resource_id)
        )
    
    async def execute(self, object_id: str):
        proposal = Proposal(
            proposal_id=f"{self.agent_id}_{object_id}",
            proposer_agent_id=self.agent_id,
            action="pick_object",
            resource_id=object_id,
            priority=self.calculate_priority(object_id),
            timestamp=time.time(),
            metadata={"location": self.position}
        )
        
        if await self.consensus.propose(proposal):
            self.robot_hal.pick_object(object_id)  # I won consensus
            return "success"
        return "deferred_to_peer"  # Another robot handling it

Benefits

1. Prevents Resource Conflicts - Multiple robots safely operate in shared spaces
2. Decentralized & Fault Tolerant - No single point of failure
3. Flexible Algorithms - RAFT for speed, PBFT for security, Gossip for scale
4. Seamless FABRIC Integration - Leverages existing identity and trust layer
5. Developer-Friendly - ~10 lines of code to add consensus to any action

Performance Targets

• Latency: <100ms consensus for small fleets (3-10 robots)
• Scalability: Gossip protocol supports 100+ robots
• Bandwidth: Minimal overhead piggybacking on FABRIC messages

Implementation Phases

If accepted, I can contribute:

1. Core consensus protocol (src/fabric/consensus.py)
2. Unit tests with simulated multi-agent scenarios
3. Integration example with existing pick_object action
4. Documentation and best practices guide
5. Performance benchmarks (RAFT vs PBFT vs Gossip)

Alternatives Considered

• Centralized Coordinator: Single point of failure, against FABRIC's decentralized vision
• Simple Priority System: Not adaptive, inefficient
• Blockchain Smart Contracts: Too slow (seconds) for real-time robotics

References

• FABRIC Architecture - OpenMind's decentralized coordination
• Raft Consensus - Understandable distributed consensus
• PBFT Paper - Byzantine fault tolerance

---

Author: @lau90eth
Related: Bounty #365 (4 sensor plugins contribution)

[Desire193]

Fabulous

[RMeireza]

Consider adding explicit support for proposal expiration or revocation in fast-changing environments (e.g., a moving object no longer needs pickup).