Distributed State Management

State management across multiple nodes: consensus, replication, partitioning, and consistency.

Challenges

Consistency Models

CAP Theorem Applied

For Agent State:
├── CP (Consistency + Partition Tolerance)
│   └── Financial, legal, safety-critical agents
├── AP (Availability + Partition Tolerance)  
│   └── Chat, creative, research agents
└── CA (Consistency + Availability)
    └── Single datacenter, no partitions

Consistency Levels

State Partitioning (Sharding)

Partition Strategies

class StatePartitioner:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards

    def get_shard(self, key: str) -> int:
        # Consistent hashing
        return hash(key) % self.num_shards

    # Partition by:
    # - Thread ID (conversation isolation)
    # - User ID (user data locality)
    # - Tenant ID (multi-tenant isolation)
    # - Agent ID (agent-specific state)

Shard Management

class ShardManager:
    def __init__(self, shards: list[Shard]):
        self.shards = {s.shard_id: s for s in shards}

    async def get_shard_for_thread(self, thread_id: str) -> Shard:
        shard_id = self.partitioner.get_shard(thread_id)
        return self.shards[shard_id]

    async def rebalance(self, new_num_shards: int):
        # 1. Compute new assignments
        # 2. Migrate data (online or offline)
        # 3. Update routing
        # 4. Verify consistency

Distributed Consensus

Raft for State Machine

# State machine replication via Raft
class RaftStateMachine:
    def __init__(self, node_id: str, peers: list[str]):
        self.node_id = node_id
        self.peers = peers
        self.log: list[LogEntry] = []
        self.state = "follower"
        self.current_term = 0
        self.voted_for = None

    async def propose(self, command: Command) -> Result:
        if self.state != "leader":
            raise NotLeader()

        entry = LogEntry(term=self.current_term, command=command)
        self.log.append(entry)

        # Replicate to quorum
        await self.replicate_to_quorum(entry)

        # Apply to state machine
        self.apply(entry.command)
        return Result(success=True)

When to Use Consensus

Distributed Locking

Redlock (Redis)

class DistributedLock:
    def __init__(self, redis_clients: list[redis.Redis]):
        self.clients = redis_clients
        self.quorum = len(redis_clients) // 2 + 1

    async def acquire(self, resource: str, ttl: int = 10) -> bool:
        lock_id = str(uuid4())
        start = time.time()

        acquired = 0
        for client in self.clients:
            if await client.set(f"lock:{resource}", lock_id, nx=True, ex=ttl):
                acquired += 1

        if acquired >= self.quorum:
            return True

        # Failed: release any acquired
        for client in self.clients:
            await client.eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", 1, f"lock:{resource}", lock_id)
        return False

    async def release(self, resource: str, lock_id: str):
        for client in self.clients:
            await client.eval("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", 1, f"lock:{resource}", lock_id)

When to Lock

Distributed Caching

Cache Patterns

class DistributedCache:
    def __init__(self, cache: CacheBackend, local_cache: LocalCache):
        self.remote = cache
        self.local = local_cache  # L1 cache

    async def get(self, key: str):
        # L1 check
        value = await self.local.get(key)
        if value is not None:
            return value

        # L2 check
        value = await self.remote.get(key)
        if value is not None:
            await self.local.set(key, value)
            return value
        return None

    async def set(self, key: str, value, ttl: int):
        await self.remote.set(key, value, ttl)
        await self.local.set(key, value, min(ttl, 300))  # L1 shorter TTL

Cache Invalidation

class CacheInvalidator:
    def __init__(self, cache: DistributedCache, pubsub: PubSub):
        self.cache = cache
        self.pubsub = pubsub
        await self.pubsub.subscribe("cache:invalidate", self.on_invalidate)

    async def on_invalidate(self, message):
        keys = message.data["keys"]
        for key in keys:
            await self.cache.local.delete(key)
            await self.cache.remote.delete(key)

    async def invalidate(self, keys: list[str]):
        await self.pubsub.publish("cache:invalidate", {"keys": keys})

Distributed Sessions

Session Affinity

class SessionRouter:
    def __init__(self, instances: list[Instance]):
        self.instances = instances

    def route(self, session_id: str) -> Instance:
        # Consistent hashing for session affinity
        return self.instances[hash(session_id) % len(self.instances)]

Session Migration

class SessionMigrator:
    async def migrate(self, session_id: str, from_node: Node, to_node: Node):
        # 1. Pause session on source
        await from_node.pause_session(session_id)

        # 2. Export state
        state = await from_node.export_session(session_id)

        # 3. Import to target
        await to_node.import_session(session_id, state)

        # 3. Resume on target
        await to_node.resume_session(session_id)

        # 4. Update routing
        await self.update_routing(session_id, to_node)

Eventual Consistency Patterns

Conflict Resolution

class ConflictResolver:
    def resolve(self, local: State, remote: State, strategy: str) -> State:
        if strategy == "last_write_wins":
            return local if local.updated_at > remote.updated_at else remote

        elif strategy == "merge":
            return self.merge_states(local, remote)

        elif strategy == "application_specific":
            return self.custom_merge(local, remote)

        elif strategy == "user_resolution":
            raise ConflictError(local, remote)

    def merge_states(self, local: State, remote: State) -> State:
        # CRDT-style merge for compatible types
        merged = {}
        for key in set(local.keys()) | set(remote.keys()):
            l, r = local.get(key), remote.get(key)
            if isinstance(l, dict) and isinstance(r, dict):
                merged[key] = self.merge_states(l, r)
            elif isinstance(l, list) and isinstance(r, list):
                merged[key] = list(set(l + r))  # Union
            elif isinstance(l, (int, float)) and isinstance(r, (int, float)):
                merged[key] = max(l, r)  # Or sum, depending on semantics
            else:
                merged[key] = r if r.timestamp > l.timestamp else l
        return merged

CRDTs (Conflict-free Replicated Data Types)

# G-Counter (Grow-only Counter)
class GCounter:
    def __init__(self, node_id: str):
        self.counts = defaultdict(int)
        self.node_id = node_id

    def increment(self):
        self.counts[self.node_id] += 1

    def merge(self, other: 'GCounter'):
        for node, count in other.counts.items():
            self.counts[node] = max(self.counts[node], count)

    def value(self) -> int:
        return sum(self.counts.values())

# LWW-Register (Last-Writer-Wins)
class LWWRegister:
    def __init__(self):
        self.value = None
        self.timestamp = 0
        self.node_id = None

    def set(self, value, timestamp, node_id):
        if timestamp > self.timestamp:
            self.value = value
            self.timestamp = timestamp
            self.node_id = node_id

    def merge(self, other: 'LWWRegister'):
        if other.timestamp > self.timestamp:
            self.value = other.value
            self.timestamp = other.timestamp
            self.node_id = other.node_id

Multi-Region Deployment

Active-Passive

Region A (Primary) ←→ Region B (Standby)
     │                      │
     ▼                      ▼
  Writes                  Reads
  Sync                    Async

Active-Active

Region A ←→ Region B ←→ Region C
  │           │           │
  ▼           ▼           ▼
All Read/Write (Conflict Resolution Required)

Data Residency

class ResidencyRouter:
    def __init__(self, regions: dict[str, Region]):
        self.regions = regions

    def route(self, user_id: str, data_type: str) -> Region:
        # Legal/compliance routing
        residency = self.get_residency_requirement(user_id, data_type)
        if residency:
            return self.regions[residency]

        # Latency-based
        return self.nearest_region(user_id)

Monitoring Distributed State

Key Metrics

Related Notes

• Checkpointing — Distributed checkpointing
• Event Sourcing — Distributed event sourcing
• Session Management — Distributed sessions
• Scaling Strategies — Horizontal scaling
• Reliability — Circuit breakers, failover
• Deployment Architecture — Multi-region
• Reliability — Circuit breakers
• Research Directions — CRDTs, consensus research

Sources: "Designing Data-Intensive Applications" (Kleppmann), "Distributed Systems" (Coulouris et al.), Raft paper, Redis Redlock, CRDT papers (Shapiro et al.), DynamoDB, Cassandra, CockroachDB, Spanner, etcd, Consul, distributed systems patterns.