Distributed State Management Patterns

Shared mutable state is the single biggest source of correctness bugs in multi-agent systems. When ten agents can read and write the same record concurrently, you are not building a distributed system, you are building a race condition at scale. This lesson covers the patterns that make distributed state safe: optimistic locking to detect conflicts, CRDTs to eliminate them, event sourcing to make state history queryable, and Redis transactions for atomic conditional updates.

Why Mutable Shared State Breaks at Scale

Consider the classic lost-update problem: Agent A reads a task at version 5, Agent B reads the same task at version 5. Agent A updates the priority and writes. Agent B updates the assignee and writes, overwriting Agent A’s priority change. Both agents believe they succeeded. The system silently lost data.

sequenceDiagram
    participant A as Agent A
    participant DB as State Store
    participant B as Agent B

    A->>DB: READ task#42 (v5)
    B->>DB: READ task#42 (v5)
    A->>DB: WRITE priority=HIGH (expected v5 → v6)
    Note over DB: Version now 6
    B->>DB: WRITE assignee=bob (expected v5 → CONFLICT!)
    DB-->>B: 409 Conflict, re-read and retry
    B->>DB: READ task#42 (v6)
    B->>DB: WRITE assignee=bob (expected v6 → v7)

    style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style DB fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style B fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Optimistic locking resolves this by making every write conditional. The agent reads a version number with the record, applies its change locally, and then writes back with a precondition: “only commit if the version is still what I read.” The database increments the version atomically; any concurrent writer with a stale version receives a conflict error and must retry.

import asyncpg
from dataclasses import dataclass

@dataclass
class Task:
    id: str
    version: int
    priority: str
    assignee: str | None

async def update_task_priority(
    pool: asyncpg.Pool,
    task_id: str,
    new_priority: str,
    max_retries: int = 5,
) -> Task:
    """
    Update task priority using optimistic locking.
    We retry on conflict rather than holding a lock, this keeps
    throughput high even with many concurrent agents because we
    never block readers or other writers.
    """
    for attempt in range(max_retries):
        async with pool.acquire() as conn:
            # Read current state including version
            row = await conn.fetchrow(
                "SELECT id, version, priority, assignee FROM tasks WHERE id = $1",
                task_id,
            )
            if row is None:
                raise ValueError(f"Task {task_id} not found")

            current_version = row["version"]

            # Conditional update: only succeeds if version hasn't changed.
            # The WHERE clause on version is the entire safety mechanism.
            updated = await conn.fetchrow(
                """
                UPDATE tasks
                SET priority = $1, version = version + 1
                WHERE id = $2 AND version = $3
                RETURNING id, version, priority, assignee
                """,
                new_priority, task_id, current_version,
            )

            if updated is not None:
                return Task(**updated)  # Commit succeeded

        # Conflict detected: brief random backoff before retry
        # to reduce collision probability among competing agents
        await asyncio.sleep(0.05 * (2 ** attempt) * random.random())

    raise RuntimeError(f"Failed to update task {task_id} after {max_retries} retries")

CRDT-Based Collaborative Agent State

Optimistic locking requires a retry loop. For high-contention state, a shared tag set, a presence map, a vote tally, a better approach is a Conflict-free Replicated Data Type (CRDT). CRDTs are designed so that any two replicas can merge independently and always converge to the same value, regardless of the order merges are applied.

flowchart TD
    A["Agent A\nslot[A] = 3"] --> M["Merge\nmax(slot[A], slot[B], slot[C])"]
    B["Agent B\nslot[B] = 7"] --> M
    C["Agent C\nslot[C] = 2"] --> M
    M --> T["Total = 3 + 7 + 2 = 12"]

    style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style B fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style C fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style M fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style T fill:#f0fdf9,stroke:#0D9488,color:#0F172A

A G-Counter is the simplest CRDT: each agent owns one slot in a vector and only ever increments its own slot. Merging two replicas takes the element-wise maximum. Because agents never write to each other’s slots, there are no conflicts, ever. The same principle scales to more complex types: OR-Sets for collaborative tag management, LWW-Registers for last-writer-wins fields, and RGA for collaborative text editing.

Event Sourcing as the Source of Truth

Rather than storing current state as a mutable row, event sourcing records every state transition as an immutable event. Current state is always a projection, computed by replaying events in sequence. This is ideal for multi-agent systems because: (1) the event log is append-only, eliminating write conflicts at the storage level; (2) every agent action is permanently auditable; and (3) agents can subscribe to the event stream to react to changes made by other agents.

flowchart TD
    E1["OrderCreated\nt=0, amount=500"] --> P["Projection\n(replay in order)"]
    E2["ItemAdded\nt=1, sku=ABC"] --> P
    E3["PriorityEscalated\nt=2, by=AgentX"] --> P
    E4["AssigneeChanged\nt=3, to=AgentY"] --> P
    P --> S["Current State\namount=500, sku=ABC\npriority=HIGH, assignee=AgentY"]

    style E1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style E2 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style E3 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style E4 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style P fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style S fill:#f0fdf9,stroke:#0D9488,color:#0F172A

Redis WATCH/MULTI/EXEC for Atomic State Updates

When agents share ephemeral coordination state, work queues, lease tables, rate-limit counters, Redis is the right store. Redis is single-threaded: every command executes atomically. But multi-step operations (read, modify, write) need the WATCH/MULTI/EXEC transaction pattern to remain atomic across the gap between read and write.

import redis.asyncio as aioredis

async def claim_work_item(
    redis: aioredis.Redis,
    queue_key: str,
    agent_id: str,
) -> str | None:
    """
    Atomically pop a work item and record which agent claimed it.
    WATCH + MULTI/EXEC ensures no other agent claims the same item
    between our read and our write: without holding a blocking lock.
    """
    async with redis.pipeline() as pipe:
        while True:
            try:
                # WATCH makes the subsequent EXEC fail if this key changes
                await pipe.watch(queue_key)

                # Read outside the transaction (WATCH is active)
                item = await pipe.lindex(queue_key, 0)
                if item is None:
                    await pipe.reset()
                    return None  # Queue is empty

                # Begin the atomic block
                pipe.multi()
                pipe.lpop(queue_key)              # Remove from queue
                pipe.hset("claims", item, agent_id)  # Record ownership
                pipe.expire("claims", 3600)       # Auto-expire stale claims

                # EXEC succeeds only if queue_key was not modified since WATCH
                await pipe.execute()
                return item.decode()

            except aioredis.WatchError:
                # Another agent modified the queue: retry from the top
                # This is the optimistic path: cheap when contention is low
                continue

Partitioning State by Agent Type

At enterprise scale, keeping all agents writing to a single state namespace creates a hot spot. Partition state by agent type or domain: state:research:{task_id}, state:writer:{task_id}, state:validator:{task_id}. Each partition has one logical owner (the agent type responsible for it) and is read-only for all other agent types. This eliminates most write conflicts at the architectural level, the same principle as database sharding applied to agent state.

For state that must be shared across types, use the event-sourcing pattern above: agents publish domain events, and any agent that needs cross-domain state builds a local read-only projection by consuming those events. This keeps writes local and reads eventually consistent, which is the right trade-off for most enterprise workloads.

State Migration Strategies

When agent logic evolves and requires schema changes to stored state, never mutate state in place. Instead: version your event schema explicitly ("event_type": "TaskPriorityChanged/v2"), write a migration agent that reads old-format events and produces new-format projections, and run old and new agents in parallel during the rollout window. Once all agents are on the new schema, the migration agent can be decommissioned. This pattern, expand/migrate/contract, is the same discipline used for zero-downtime database migrations and applies equally to agent state.

In the next lesson, we move from ephemeral coordination state to persistent long-term memory, building a PostgreSQL-backed episodic memory system with vector search so agents can recall relevant context across sessions.