Simulating Chaos: Property-Based Tests for Agent Swarm Consensus Protocols

When you have three autonomous agents voting on whether to commit a transaction, the happy path is easy. All three say yes, the transaction commits. But what happens when one agent's network link stutters, another's local clock drifts, and the third receives a conflicting proposal mid-election? Unit tests, with their deterministic setup and linear execution, will never catch the cascading failure that follows.

I've spent the last year building agent swarms that coordinate via lightweight consensus protocols—no Paxos, no Raft, just simple majority voting with timeouts and retries. The first time I ran a swarm of five agents in production, a split-brain scenario caused duplicate payments. The logs showed each agent believed it had a valid majority. The bug was a race condition in the commit phase that only manifested under specific interleavings of message delivery.

This is where property-based testing (PBT) shines. Instead of writing example-based tests ("agent A sends yes, B sends yes, C sends no → commit"), you define invariants that must hold for all possible sequences of events. Then a library like Hypothesis generates random schedules, message drops, and clock skews to try to break those invariants. When it finds a counterexample, it shrinks it to the minimal failing sequence.

The Consensus Invariant

For any consensus protocol, the core invariant is: no two agents ever commit conflicting decisions. In a payment system, that means at most one agent authorizes a given payment ID. Formally:

def consensus_invariant(history: list[Event]) -> bool:
    commits = [e for e in history if e.type == 'COMMIT']
    for c1 in commits:
        for c2 in commits:
            if c1.agent_id != c2.agent_id and c1.value == c2.value:
                return False
    return True

That's the safety property. Liveness is another invariant: if a majority of agents are healthy and the network eventually delivers messages, then a proposed value should eventually be committed.

def liveness_invariant(history: list[Event]) -> bool:
    proposals = [e for e in history if e.type == 'PROPOSE']
    for p in proposals:
        if p.value not in [c.value for c in history if c.type == 'COMMIT']:
            # Check if a majority was healthy
            healthy_agents = {e.agent_id for e in history if e.type == 'HEARTBEAT'}
            if len(healthy_agents) >= 3:  # assuming 5 agents
                return False
    return True

Building the Simulation

A property-based test needs a model of the system. For agent swarms, the model includes:

• A set of agents, each with a state machine (IDLE, VOTING, COMMITTED, FAILED).
• A network that can delay, drop, reorder, or duplicate messages.
• A clock that can tick at variable speeds per agent.

Here's a simplified simulator in Python using the hypothesis library:

from hypothesis import given, strategies as st
from enum import Enum, auto

class AgentState(Enum):
    IDLE = auto()
    VOTING = auto()
    COMMITTED = auto()
    FAILED = auto()

class Message:
    def __init__(self, sender, msg_type, value, round):
        self.sender = sender
        self.msg_type = msg_type  # 'PROPOSE', 'VOTE', 'COMMIT'
        self.value = value
        self.round = round

def simulate(agents, network_events, clock_skews):
    # agents: dict of {agent_id: AgentState}
    # network_events: list of (time, from, to, message)
    # clock_skews: dict of {agent_id: offset}
    history = []
    # ... event loop with priority queue ...
    return history

The key is to generate random sequences that stress the protocol. Hypothesis lets you compose strategies:

@st.composite
def consensus_scenario(draw):
    num_agents = draw(st.integers(min_value=3, max_value=7))
    agents = {i: AgentState.IDLE for i in range(num_agents)}
    
    # Generate random message delays (0-100ms)
    delays = draw(st.lists(st.floats(min_value=0, max_value=100), 
                           min_size=num_agents, max_size=num_agents))
    
    # Generate random message drops (10% chance)
    drop_prob = 0.1
    
    # Generate random clock skews (-50 to +50 ms)
    skews = draw(st.lists(st.floats(min_value=-50, max_value=50),
                          min_size=num_agents, max_size=num_agents))
    
    # Generate a sequence of proposals
    proposals = draw(st.lists(st.tuples(st.integers(0, num_agents-1), 
                                        st.text(min_size=1, max_size=10)),
                              min_size=1, max_size=10))
    
    return agents, delays, drop_prob, skews, proposals

Running the Test

With the scenario generator, you define a test that checks invariants:

@given(consensus_scenario())
def test_consensus_safety(agents, delays, drop_prob, skews, proposals):
    network = Network(delays, drop_prob)
    clock = Clock(skews)
    swarm = Swarm(agents, network, clock)
    
    for proposer, value in proposals:
        swarm.propose(proposer, value)
        swarm.tick(until_idle=True)
    
    history = swarm.history
    assert consensus_invariant(history), f"Safety violated: {history}"
    assert liveness_invariant(history), f"Liveness violated: {history}"

When I first ran this against my naive consensus implementation, Hypothesis found a counterexample within seconds. The minimal failing case: three agents, one proposal, a message from agent 0 to agent 1 was delayed by 200ms while agent 2's clock was 50ms ahead. Agent 2 timed out, started a new round, and committed a conflicting value before agent 1 received the original vote.

The fix was to add a unique round number to each proposal and require that agents only vote for the highest round they've seen. This is essentially the Paxos trick, but encoded as a simple integer comparison.

Beyond Safety: Liveness Under Duress

Safety is necessary but not sufficient. A protocol that never commits anything is safe but useless. Liveness properties are harder to test because they require eventual guarantees. Hypothesis can still help by bounding the simulation time and checking that a commit occurs within that bound.

def test_liveness_bounded():
    # Run many random scenarios, each with a max tick count
    for _ in range(1000):
        scenario = random_scenario()
        swarm = Swarm(*scenario)
        for _ in range(100):  # max ticks
            swarm.tick()
            if any(a.state == AgentState.COMMITTED for a in swarm.agents):
                break
        else:
            # No commit within 100 ticks
            if majority_healthy(swarm):
                raise AssertionError("Liveness failure")

This revealed that my retry mechanism had a bug: when an agent's vote message was dropped, it would retry indefinitely without ever receiving a quorum, because other agents had already moved to the next round. The fix was to add a random backoff and allow agents to re-enter voting if they missed the window.

Integrating with Your CI

Property-based tests are slower than unit tests, so they belong in a separate CI stage that runs nightly or on demand. I keep them in a tests/property/ directory and run them with pytest --hypothesis-profile=ci which sets a higher max examples and longer timeout.

# .github/workflows/property-tests.yml
- name: Run property-based tests
  run: |
    cd tests/property
    pytest --hypothesis-profile=ci --timeout=300

Hypothesis also supports stateful testing, where you model the system as a state machine and let the library generate sequences of operations. That's ideal for testing consensus protocols because it can explore interleavings that you'd never think to write manually.

The Payoff

Since adding property-based tests, I've caught three consensus bugs before deployment. One was a subtle integer overflow in the round counter (after 2^32 rounds, it wrapped to zero, causing a split-brain). Another was a race condition where a delayed COMMIT message from a previous round was applied to the current round. Both would have been nearly impossible to find with unit tests.

Property-based testing doesn't replace integration or stress tests—it complements them. It's a systematic way to explore the chaotic space of distributed agent behavior. If your swarm handles money, votes, or access tokens, you owe it to yourself to simulate the chaos before it happens for real.