# How to Handle State Conflicts with StateConflictError in aisuite

> Learn how to handle StateConflictError in aisuite. Prevent data loss by understanding optimistic concurrency and managing stale revision numbers effectively.

- Repository: [Andrew Ng/aisuite](https://github.com/andrewyng/aisuite)
- Tags: how-to-guide
- Published: 2026-07-28

---

**`StateConflictError` is raised when you attempt to save agent state using a stale revision number, preventing concurrent writes from silently overwriting each other in aisuite's optimistic concurrency model.**

aisuite provides persistent state management for AI agent conversations through pluggable state stores. When building distributed applications or multi-threaded systems, multiple processes may attempt to update the same conversation thread simultaneously. The framework detects these collisions using revision-based optimistic locking, raising `StateConflictError` in [`aisuite/agents/state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py) when a write operation targets an outdated version of the state.

## What Triggers StateConflictError

The error originates in the `_assert_revision` helper function located in [`aisuite/agents/state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py). Before persisting any state, the store compares the revision number you provide against the current revision stored for that thread:

```python
def _assert_revision(thread_id, current_revision, expected_revision):
    if expected_revision is None:
        return
    if current_revision != expected_revision:
        raise StateConflictError(
            f"State revision conflict for {thread_id!r}: "
            f"expected {expected_revision}, found {current_revision}."
        )

```

When `expected_revision` (the version you last read) does not match `current_revision` (the version currently stored), aisuite raises `StateConflictError`. This check runs for every non-initial save operation where a revision number is explicitly provided.

## Where StateConflictError Occurs

All three state store implementations in aisuite utilize the `_assert_revision` check to enforce data integrity:

- **InMemoryStateStore** – The default in-memory backend validates revisions in `save_state()` before updating its internal dictionary.
- **FileStateStore** – The file-based persistence layer performs the check in `save_state()` prior to writing JSON files to disk.
- **PostgreSQLStateStore** – The production PostgreSQL backend implements identical logic in [`aisuite/agents/postgres_state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/postgres_state_store.py), using database transactions with optimistic locking rather than table locks.

This consistent behavior across backends allows you to swap storage implementations without changing your conflict-handling logic.

## The Optimistic Concurrency Workflow

Handling state conflicts follows a standard read-modify-retry pattern. To avoid `StateConflictError`, your application should:

1. **Load** the current `StoredRunState` (or initialize if none exists).
2. **Capture** the `revision` property from the loaded state.
3. **Modify** the `RunState` object (e.g., appending messages).
4. **Save** the updated state, passing the captured revision as the `revision` parameter.
5. **Catch** `StateConflictError` and restart from step 1 using the latest state if the error occurs.

This pattern eliminates the need for distributed locks while guaranteeing that no update is lost.

## Practical Implementation Examples

### Basic Conflict Handling with InMemoryStateStore

The following example demonstrates detecting and recovering from a conflict using the in-memory store:

```python
import aisuite as ai

store = ai.InMemoryStateStore()
thread_id = "thread/user:42"

# Load existing state or create initial state

saved = store.save_state(thread_id, ai.RunState(agent_name="assistant"))
revision = saved.revision

# Modify state

saved.state.add_user_message("Hello, world!")

# Attempt to persist with version check

try:
    store.save_state(thread_id, saved.state, revision=revision)
except ai.StateConflictError:
    # State changed elsewhere - fetch latest and retry

    latest = store.load_state(thread_id)
    latest.state.add_user_message("Hello, world!")
    store.save_state(thread_id, latest.state, revision=latest.revision)

```

### Automatic Retry Logic with FileStateStore

For file-based persistence, implement recursive or loop-based retry logic to handle transient conflicts:

```python
import aisuite as ai
from pathlib import Path

store = ai.FileStateStore(Path(".aisuite/state"))
thread_id = "thread/user:7"

def update_state_with_retry(max_attempts=3):
    stored = store.load_state(thread_id)
    
    if stored is None:
        # Initial write requires no revision check

        state = ai.RunState(agent_name="assistant")
        store.save_state(thread_id, state)
        return
    
    try:
        # Apply changes to loaded state

        stored.state.add_user_message("Processing update")
        store.save_state(thread_id, stored.state, revision=stored.revision)
    except ai.StateConflictError:
        if max_attempts > 0:
            # Retry with fresh state

            update_state_with_retry(max_attempts - 1)
        else:
            raise

update_state_with_retry()

```

### HTTP 409 Responses in Web Applications

When exposing aisuite through REST APIs, map `StateConflictError` to HTTP 409 Conflict status codes:

```python
from fastapi import FastAPI, HTTPException
import aisuite as ai

app = FastAPI()
state_store = ai.FileStateStore(".aisuite/state")

@app.post("/threads/{thread_id}/messages")
def add_message(thread_id: str, content: str):
    stored = state_store.load_state(thread_id)
    
    if stored is None:
        stored = ai.StoredRunState(
            thread_id=thread_id,
            state=ai.RunState(agent_name="assistant"),
            revision=0,
            created_at=ai.utils.now(),
            updated_at=ai.utils.now(),
        )
    
    revision = stored.revision
    stored.state.add_user_message(content)
    
    try:
        state_store.save_state(thread_id, stored.state, revision=revision)
    except ai.StateConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    
    return {"status": "saved", "revision": revision + 1}

```

## Best Practices for State Conflict Resolution

- **Reload before retrying**: Always call `load_state()` to fetch the current revision before reapplying changes. Reusing the old state object will trigger another conflict.
- **Implement idempotency**: Design your state modifications so that applying them twice produces the same result, preventing duplicate messages during retries.
- **Limit retry attempts**: Use exponential backoff with a maximum retry count to prevent infinite loops during high-contention scenarios.
- **Pass `None` for initial writes**: When creating a new thread that has no existing state, omit the `revision` parameter or pass `None` to bypass the revision check.

## Summary

- **`StateConflictError`** signals an optimistic concurrency failure when saving agent state in aisuite.
- The error is raised by **`_assert_revision`** in [`aisuite/agents/state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py) whenever the provided revision does not match the stored revision.
- All backends—including **InMemoryStateStore**, **FileStateStore**, and **PostgreSQLStateStore**—implement identical conflict detection.
- Handle conflicts by catching the error, reloading the latest state via `load_state()`, and retrying the save operation with the updated revision.
- Map `StateConflictError` to **HTTP 409** in web applications to indicate concurrent modification conflicts to clients.

## Frequently Asked Questions

### What is StateConflictError in aisuite?

`StateConflictError` is a `RuntimeError` subclass defined in [`aisuite/agents/state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py). It indicates that an attempt to save agent state failed because the thread was modified by another process between the time you read the state and the time you attempted to write it, preventing lost updates in concurrent environments.

### How does aisuite detect state conflicts?

aisuite detects conflicts through revision-based optimistic concurrency control. Each saved state carries a monotonically increasing revision number. When you call `save_state()` with a `revision` parameter, the store compares it against the current revision using `_assert_revision()`. If they differ, the operation aborts with `StateConflictError`.

### Should I catch StateConflictError in production applications?

Yes, production applications using shared state stores must catch `StateConflictError` to handle race conditions gracefully. Implement a retry mechanism that reloads the latest state, re-applies your business logic, and attempts the save again. For read-heavy workloads, this approach outperforms pessimistic locking while maintaining data integrity.

### Does the PostgreSQL implementation use the same conflict logic?

Yes, [`aisuite/agents/postgres_state_store.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/postgres_state_store.py) implements the same `_assert_revision` logic as the in-memory and file stores. While it uses PostgreSQL as the backing database, it performs the revision check in application code rather than using `SELECT FOR UPDATE`, maintaining consistency across all aisuite storage backends.