Collaborative State & Conflict Resolution
When a single agent works on a task, state is simple. One context window, one execution trace, and one scratchpad of accumulated facts create a sequential flow. The agent reads its own state, modifies it, and moves on.
Multi-agent systems destroy that simplicity. The moment two agents work on the same task — editing the same document, updating the same database record, contributing to the same research report — they share mutable state. And shared mutable state, in any system, is where bugs live. One agent reads a document, plans edits based on what it sees, and writes those edits back. Meanwhile, another agent has done the same thing, and its write overwrites the first agent's changes. Both agents proceed under the false assumption that their changes survived, and the final output becomes a mashup that diverges from both intentions.
It is the same category of concurrency bug that has plagued distributed systems for decades. Agents add nondeterministic read and write behavior, which makes runtime conflict detection, prevention, and resolution essential.
What Agents Share #
Before choosing a concurrency strategy, you need to understand what kinds of state agents actually share. Different state types demand different approaches.
┌──────────────────────────────────────────────────────────────┐
│ Shared State in Multi-Agent Systems │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Shared Document │ │ Task Board │ │
│ │ (report, code, │ │ (who is doing │ │
│ │ plan draft) │ │ what, status) │ │
│ └──────────────────┘ └──────────────────┘ │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Knowledge Base │ │ Execution Log │ │
│ │ (facts, findings│ │ (what happened, │ │
│ │ conclusions) │ │ results) │ │
│ └──────────────────┘ └──────────────────┘ │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Configuration │ │ External State │ │
│ │ (parameters, │ │ (database rows, │ │
│ │ thresholds) │ │ API resources) │ │
│ └──────────────────┘ └──────────────────┘ │
└──────────────────────────────────────────────────────────────┘
Shared documents are the most common case. Two agents collaborating on a research report, a code file, or a project plan both need to read the current version and write changes. This is the hardest state to share safely because edits are semantic — one agent might restructure a paragraph while another inserts a sentence, and merging those changes requires understanding the content.
Task boards track which agent is working on what. These are simpler — the state is structured metadata (task ID, assignee, status), and conflicts are less damaging. If two agents both try to claim the same task, you pick a winner and the loser grabs a different one.
Knowledge bases are append-mostly. Agents add findings, facts, and conclusions as they work. Independent additions rarely collide, though isolated evidence can produce contradictions such as "revenue grew 12%" and "revenue grew 8%."
External state — database rows, API resources, file system objects — is the most dangerous because side effects are irreversible. If two agents both try to update a customer record or deploy a service, the conflict plays out in the real world.
Naive Sharing #
The simplest approach — and the one used for multi-agent prototypes — is to let all agents read and write shared state freely through tools.
# Naive shared state: a dictionary every agent can access
import json
shared_state = {}
def read_state(key: str) -> str:
return json.dumps(shared_state.get(key, None))
def write_state(key: str, value: str) -> str:
shared_state[key] = json.loads(value)
return "OK"
# Both agents get these tools
research_agent = Agent(
tools=[read_state_tool, write_state_tool, search_tool],
system_prompt="Research the topic. Store findings in shared state.",
)
analysis_agent = Agent(
tools=[read_state_tool, write_state_tool],
system_prompt="Analyze the research findings. Update conclusions in shared state.",
)
This works when agents execute sequentially — one finishes before the next starts. Concurrent execution for latency or throughput introduces classic concurrency bugs:
Lost updates. Agent A reads findings = ["fact1"], appends "fact2", and writes ["fact1", "fact2"]. Agent B reads the same original ["fact1"], appends "fact3", and writes ["fact1", "fact3"]. Agent B's write silently drops "fact2".
Stale reads. Agent A reads a document, spends 30 seconds planning edits, and writes its changes. During those 30 seconds, Agent B modifies the same document, leaving Agent A's write based on an obsolete version.
Inconsistent state. Agent A updates two related fields — total_cost and line_items — but Agent B reads between the two writes, seeing the new line items with the old total.
These cases are the expected behavior of uncoordinated concurrent writes. The fix requires choosing a concurrency model.
Optimistic Concurrency Control #
Optimistic concurrency assumes conflicts are rare and detects them at write time. Each piece of state carries a version number. When an agent reads state, it gets the current version. When it writes, it includes the version it read. If the version has changed since the read, the write is rejected and the agent must re-read and retry.
from dataclasses import dataclass, field
import time
from typing import Any
@dataclass
class VersionedValue:
value: Any
version: int = 0
last_modified_by: str = ""
timestamp: float = field(default_factory=time.time)
class VersionedStore:
"""Shared state store with optimistic concurrency control."""
def __init__(self):
self._data: dict[str, VersionedValue] = {}
def read(self, key: str) -> tuple[Any, int]:
entry = self._data.get(key)
if entry is None:
return None, 0
return entry.value, entry.version
def write(
self,
key: str,
value: Any,
expected_version: int,
writer_id: str,
) -> bool:
"""
Write only if current version matches expected_version.
Returns True on success, False on conflict.
"""
current = self._data.get(key)
current_version = current.version if current else 0
if current_version != expected_version:
return False # Conflict — someone else wrote first
self._data[key] = VersionedValue(
value=value,
version=current_version + 1,
last_modified_by=writer_id,
)
return True
The agent's tools wrap this store with conflict handling:
def make_versioned_tools(store: VersionedStore, agent_id: str) -> list:
version_cache: dict[str, int] = {}
async def read(key: str) -> str:
value, version = store.read(key)
version_cache[key] = version
return json.dumps({"value": value, "version": version})
async def write(key: str, value: str) -> str:
expected = version_cache.get(key, 0)
parsed = json.loads(value)
success = store.write(key, parsed, expected, agent_id)
if success:
version_cache[key] = expected + 1
return "Write successful"
# Conflict: return the current value so the agent can retry
current_value, current_version = store.read(key)
version_cache[key] = current_version
return json.dumps({
"error": "Conflict: state was modified by another agent",
"current_value": current_value,
"current_version": current_version,
})
return [
make_tool("read_state", "Read a value from shared state", ..., read),
make_tool("write_state", "Write a value to shared state", ..., write),
]
When a conflict occurs, the tool returns the current value so the model can see what changed and decide what to do. The model might merge its intended changes with the new state, abandon its update, or try a different approach. Agents turn retry logic from a mechanical "read-modify-write" loop into a semantic decision about how to resolve the conflict.
Trade-offs: Optimistic concurrency is simple, uses version checks in place of locking infrastructure, and works well when conflicts are infrequent. When conflicts are frequent — many agents writing to the same key — agents spend most of their time retrying, which wastes tokens and adds latency.
Event Sourcing #
Event sourcing stores the sequence of changes that produced the current state. Each agent appends events to an immutable log, and replaying that log derives the current state.
@dataclass
class StateEvent:
event_id: str
agent_id: str
timestamp: float
event_type: str # "set", "append", "delete", "patch"
key: str
payload: Any
class EventSourcedStore:
"""
State derived from an append-only event log.
No writes can conflict because every event is an append.
"""
def __init__(self):
self._events: list[StateEvent] = []
def append_event(self, event: StateEvent) -> None:
self._events.append(event)
def get_current_state(self) -> dict[str, Any]:
state: dict[str, Any] = {}
for event in self._events:
if event.event_type == "set":
state[event.key] = event.payload
elif event.event_type == "append":
if event.key not in state:
state[event.key] = []
state[event.key].append(event.payload)
elif event.event_type == "delete":
state.pop(event.key, None)
elif event.event_type == "patch":
if event.key in state and isinstance(state[event.key], dict):
state[event.key].update(event.payload)
return state
def get_events_since(self, after_event_id: str) -> list[StateEvent]:
found = False
result = []
for event in self._events:
if found:
result.append(event)
if event.event_id == after_event_id:
found = True
return result
Event sourcing preserves every update because agents append to the log. If two agents both add findings, the log retains both, and reconciliation can focus on contradictory events.
async def reconcile_findings(
store: EventSourcedStore,
model: "ModelClient",
) -> str:
"""
Use an LLM to reconcile contradictory events in the log.
"""
state = store.get_current_state()
findings = state.get("findings", [])
if len(findings) < 2:
return json.dumps(findings)
response = await model.chat(
system="""You are a fact reconciler. Given a list of findings from
multiple research agents, identify contradictions, resolve them
where possible, and produce a consolidated list.""",
messages=[{"role": "user", "content": json.dumps(findings)}],
)
return response.text
This is a powerful pattern: let agents work freely and accumulate events, then use a separate reconciliation step (which can itself be an LLM call) to resolve conflicts after the fact. It trades real-time consistency for throughput: lock-free progress allows a temporarily contradictory state until reconciliation runs.
Trade-offs: Event logs grow indefinitely. For long-running tasks, you need compaction — periodically snapshot the current state and truncate old events. Replaying a thousand events to compute current state is slow; combining event sourcing with periodic snapshots (as discussed in long-running agents) keeps reconstruction fast.
CRDTs for Agent State #
Conflict-free Replicated Data Types (CRDTs) are data structures that multiple actors can modify independently and merge automatically. They restrict operations to ones that are commutative, associative, and idempotent, making every application order equivalent.
For agent systems, a few CRDT types are especially useful:
Grow-only sets (G-Sets) support additive operations. Two agents adding different research findings to a G-Set will always merge cleanly. This fits knowledge bases, finding lists, and fact collections.
Last-writer-wins registers (LWW-Registers) — each write carries a timestamp, and the most recent write wins. Simple but lossy — earlier writes are silently discarded.
Observed-remove sets (OR-Sets) — agents can both add and remove items, with removal tracked per-add so that concurrent add-and-remove pairs resolve deterministically.
from dataclasses import dataclass, field
@dataclass
class GrowOnlySet:
"""A set that supports only adds. Merges are always conflict-free."""
items: set = field(default_factory=set)
def add(self, item: str) -> None:
self.items.add(item)
def merge(self, other: "GrowOnlySet") -> "GrowOnlySet":
return GrowOnlySet(items=self.items | other.items)
def query(self) -> set:
return self.items.copy()
@dataclass
class LWWRegister:
"""Last-writer-wins register with deterministic tie-breaking."""
value: Any = None
timestamp: float = 0.0
writer_id: str = ""
def set(self, value: Any, timestamp: float, writer_id: str) -> None:
if (timestamp, writer_id) > (self.timestamp, self.writer_id):
self.value = value
self.timestamp = timestamp
self.writer_id = writer_id
def merge(self, other: "LWWRegister") -> "LWWRegister":
winner = max(
(self, other),
key=lambda register: (register.timestamp, register.writer_id),
)
return LWWRegister(
value=winner.value,
timestamp=winner.timestamp,
writer_id=winner.writer_id,
)
class AgentCRDTState:
"""
Shared state using CRDTs for conflict-free concurrent updates.
Each agent holds a local replica and periodically merges with peers.
"""
def __init__(self):
self.findings = GrowOnlySet()
self.conclusions = LWWRegister()
self.task_claims: dict[str, LWWRegister] = {}
def add_finding(self, finding: str) -> None:
self.findings.add(finding)
def set_conclusion(
self,
conclusion: str,
timestamp: float,
agent_id: str,
) -> None:
self.conclusions.set(conclusion, timestamp, agent_id)
def claim_task(self, task_id: str, agent_id: str, timestamp: float) -> None:
if task_id not in self.task_claims:
self.task_claims[task_id] = LWWRegister()
self.task_claims[task_id].set(agent_id, timestamp, agent_id)
def merge(self, other: "AgentCRDTState") -> "AgentCRDTState":
merged = AgentCRDTState()
merged.findings = self.findings.merge(other.findings)
merged.conclusions = self.conclusions.merge(other.conclusions)
all_task_ids = set(self.task_claims) | set(other.task_claims)
for tid in all_task_ids:
mine = self.task_claims.get(tid, LWWRegister())
theirs = other.task_claims.get(tid, LWWRegister())
merged.task_claims[tid] = mine.merge(theirs)
return merged
Agent A (replica) Agent B (replica)
findings: {f1, f2} findings: {f1, f3}
conclusion: "X" @t=5 conclusion: "Y" @t=7
merge
│
▼
Merged state:
findings: {f1, f2, f3} ◄── union (G-Set)
conclusion: "Y" @t=7 ◄── latest wins (LWW)
CRDTs are attractive because merges are automatic and always succeed, with a deliberately constrained operation set. Free-form edits such as "edit paragraph 3 of this document" require semantic merging because application order changes their meaning. CRDTs work best for structured, additive state — facts, tags, status fields, counters.
Semantic Conflict Resolution #
Agent conflicts often carry meaning beyond structure. Two agents write different conclusions based on different reasoning, and a structural merge (pick the latest, union the sets) might produce a technically consistent state that is logically incoherent.
@dataclass
class ConflictRecord:
key: str
values: list[dict] # Each has "value", "agent_id", "reasoning"
resolution: str | None = None
resolved_by: str | None = None
async def resolve_semantic_conflict(
conflict: ConflictRecord,
model: "ModelClient",
) -> str:
"""
Use an LLM to resolve a semantic conflict between
agent outputs by evaluating their reasoning.
"""
conflict_description = "\n\n".join(
f"Agent '{v['agent_id']}' concluded: {v['value']}\n"
f"Reasoning: {v['reasoning']}"
for v in conflict.values
)
response = await model.chat(
system="""You are a conflict resolver. Multiple agents have produced
conflicting conclusions about the same question. Evaluate each
agent's reasoning, identify which is better supported, and produce
a single resolved conclusion. If both have merit, synthesize them.
Explain your resolution briefly.""",
messages=[{"role": "user", "content": (
f"Conflict on '{conflict.key}':\n\n{conflict_description}"
)}],
)
conflict.resolution = response.text
conflict.resolved_by = "resolver_model"
return response.text
This is the agent-specific twist on conflict resolution: when the conflicting values are natural language conclusions, a language model is often the best arbiter. It can read both agents' reasoning, evaluate the evidence, and produce a semantically coherent synthesis beyond the reach of mechanical merge strategies.
The design pattern is: use structural conflict resolution (CRDTs, versioning, last-writer-wins) for state that has a clear technical merge semantic, and use LLM-based resolution for state that requires understanding the content.
Ownership and Partitioning #
The simplest way to prevent conflicts is single-writer ownership. Assigning each piece of state to exactly one writing agent serializes updates at the source.
@dataclass
class StatePartition:
key_prefix: str
owner_agent_id: str
readable_by: list[str] = field(default_factory=list)
class PartitionedStore:
"""
State store where each key range is owned by one agent.
Other agents can read but not write.
"""
def __init__(self, partitions: list[StatePartition]):
self.partitions = {p.key_prefix: p for p in partitions}
self._data: dict[str, Any] = {}
def _find_partition(self, key: str) -> StatePartition | None:
# Prefer the most specific prefix when partitions overlap.
for prefix in sorted(self.partitions, key=len, reverse=True):
if key.startswith(prefix):
return self.partitions[prefix]
return None
def read(self, key: str, reader_id: str) -> Any:
partition = self._find_partition(key)
if partition is None:
return None
if reader_id != partition.owner_agent_id and reader_id not in partition.readable_by:
raise PermissionError(f"Agent '{reader_id}' cannot read key '{key}'")
return self._data.get(key)
def write(self, key: str, value: Any, writer_id: str) -> None:
partition = self._find_partition(key)
if partition is None:
raise KeyError(f"No partition for key '{key}'")
if writer_id != partition.owner_agent_id:
raise PermissionError(
f"Agent '{writer_id}' cannot write to key '{key}' "
f"(owned by '{partition.owner_agent_id}')"
)
self._data[key] = value
# Research agent owns "research.*", analysis agent owns "analysis.*"
store = PartitionedStore([
StatePartition("research.", "research_agent", readable_by=["analysis_agent"]),
StatePartition("analysis.", "analysis_agent", readable_by=["research_agent"]),
])
Partitioning mirrors how multi-agent systems already divide responsibilities. If each agent has a distinct job, it makes sense that each agent owns the state associated with that job. The research agent writes findings; the analysis agent reads findings and writes conclusions. Their separate write domains eliminate overlap.
The limitation is cross-cutting state — state that legitimately needs to be written by multiple agents. A shared task board, a collaborative document, a global status tracker. For these, you need one of the concurrency mechanisms above. But partitioning should be the default, with shared mutable state as the exception that gets explicit handling.
Coordination Patterns #
Beyond the data structures, there are higher-level patterns for how agents coordinate their state changes.
Turn-Based Coordination #
The simplest coordination model: agents take turns. Only one agent is active at a time, and it has exclusive access to all shared state during its turn. This eliminates concurrency entirely.
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent A │ │ Agent B │ │ Agent A │ │ Agent B │
│ Turn 1 │────▶│ Turn 2 │───▶│ Turn 3 │────▶│ Turn 4 │
│ │ │ │ │ │ │ │
│ read + │ │ read + │ │ read + │ │ read + │
│ write │ │ write │ │ write │ │ write │
└─────────┘ └─────────┘ └─────────┘ └─────────┘
No concurrency → no conflicts → no complexity
Cost: total latency = sum of all turns
Turn-based coordination is what the review-and-critique pattern uses implicitly — the generator writes, the reviewer reads and provides feedback, the generator reads the feedback and writes a revision. Serialized access prevents conflicts at the cost of latency: every agent waits for every other agent. This suits workflows where agents build sequentially on each other's work, while parallelizable tasks benefit from another strategy.
Message-Passing Coordination #
Message passing gives each agent private state and a communication channel. When one agent needs to inform another, it sends a message, and the receiving agent integrates that message into its own state however it sees fit.
import asyncio
from dataclasses import dataclass
@dataclass
class AgentMessage:
sender: str
recipient: str
content: dict
message_type: str # "finding", "request", "conclusion", "conflict"
class MessageBus:
def __init__(self):
self._queues: dict[str, asyncio.Queue] = {}
def register(self, agent_id: str) -> None:
self._queues.setdefault(agent_id, asyncio.Queue())
async def send(self, message: AgentMessage) -> None:
if message.recipient in self._queues:
await self._queues[message.recipient].put(message)
async def receive(self, agent_id: str) -> AgentMessage | None:
if agent_id not in self._queues:
return None
try:
return self._queues[agent_id].get_nowait()
except asyncio.QueueEmpty:
return None
async def broadcast(self, sender: str, content: dict, msg_type: str) -> None:
for agent_id in self._queues:
if agent_id != sender:
await self.send(AgentMessage(sender, agent_id, content, msg_type))
Message passing keeps each agent's state private and internally consistent. The challenge moves to message ordering and eventual consistency — if Agent A sends two messages and Agent B processes them out of order, does the result still make sense? Most agent workloads (passing findings, requesting work, sharing conclusions) tolerate flexible ordering. Workflows with causal dependencies ("apply edit A before edit B") need sequence numbers or causal ordering.
Leader Election for Write Access #
When multiple agents need to write to the same shared resource, elect a leader. The leader holds exclusive write access for a bounded time period, does its work, and then releases the lock. Other agents retain read access while the leader serves as the sole writer.
import time
class LeaderLock:
"""Simple leader election for shared resource access."""
def __init__(self, lease_duration: float = 30.0):
self.current_leader: str | None = None
self.lease_expires: float = 0.0
self.lease_duration = lease_duration
def try_acquire(self, agent_id: str) -> bool:
now = time.monotonic()
if self.current_leader is None or now > self.lease_expires:
self.current_leader = agent_id
self.lease_expires = now + self.lease_duration
return True
return self.current_leader == agent_id
def release(self, agent_id: str) -> None:
if self.current_leader == agent_id:
self.current_leader = None
self.lease_expires = 0.0
def is_leader(self, agent_id: str) -> bool:
return (
self.current_leader == agent_id
and time.monotonic() < self.lease_expires
)
Leader election is useful when agents take turns editing a shared document or updating a shared plan. The lease duration should cover a meaningful unit of work while limiting the delay caused by a crashed agent.
Choosing a Strategy #
The right concurrency strategy depends on the collision frequency and the cost of getting it wrong.
| Strategy | Best when | Watch out for |
|---|---|---|
| Partitioned ownership | Agents have distinct responsibilities | Cross-cutting state needs special handling |
| Turn-based | Sequential workflow, low parallelism | Latency adds up across many agents |
| Optimistic concurrency | Conflicts are rare, reads vastly outnumber writes | High-conflict keys cause retry storms |
| Event sourcing | Append-heavy workloads, audit trail needed | Log growth, replay cost, reconciliation complexity |
| CRDTs | Structured additive state with distributed coordination | Operation set limited to merge-safe structures |
| Message passing | Agents are loosely coupled, private state preferred | Message ordering, eventual consistency |
| Leader election | One resource, multiple potential writers | Leader failure blocks all writers until lease expires |
For most multi-agent systems, start with partitioned ownership — give each agent its own state namespace. When you find state that genuinely needs multiple writers, reach for optimistic concurrency if conflicts are rare or event sourcing if you need an audit trail. Reserve CRDTs for specific data structures where conflict-free merges are mathematically possible. Use semantic resolution — an LLM call to reconcile — for conflicts that exceed structural strategies.
Conclusion #
Shared mutable state is the hardest problem in multi-agent systems, for the same reason it is the hardest problem in distributed systems: concurrent writes to the same data produce bugs that are hard to reproduce, hard to diagnose, and easy to miss.
The key ideas to carry forward:
- Partition first. Give each agent ownership of its own state and introduce shared mutable state only where multiple writers provide clear value. Most multi-agent tasks can be structured so that agents write to different keys and read from each other.
- Optimistic concurrency handles the common case. Version numbers on shared state catch conflicting writes with lightweight coordination. When a conflict occurs, the agent sees the current state and can reason about how to merge.
- Event sourcing trades consistency for throughput. Append-only logs eliminate write conflicts entirely but require a reconciliation step to resolve contradictory events. This reconciliation can itself be an LLM call.
- CRDTs work for structured state. Grow-only sets, last-writer-wins registers, and counters merge automatically and deterministically. Free-form text and semantic conflicts require semantic resolution.
- Semantic conflicts need semantic resolution. When two agents produce conflicting conclusions based on different reasoning, a language model is often the best judge. This is the agent-specific capability that traditional distributed systems lack.
- Coordination patterns complement data structures. Turn-based access, message passing, and leader election shape when agents access state. CRDTs, versioning, and event logs shape how conflicting accesses are resolved. Use both.