How to Handle State Conflicts with StateConflictError in aisuite

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 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. Before persisting any state, the store compares the revision number you provide against the current revision stored for that thread:

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, 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:

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:

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:

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 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. 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}