How ThreadAlreadyExistsError and StateConflictError Protect Thread Integrity in aisuite

ThreadAlreadyExistsError prevents overwriting existing conversation threads, while StateConflictError blocks concurrent writes through optimistic concurrency control, ensuring immutable and consistent AI agent execution state.

The aisuite library treats every persisted agent execution as an immutable thread. To enforce this immutability guarantee and prevent race conditions in concurrent environments, the framework raises two specific exceptions: ThreadAlreadyExistsError when attempting to recreate an existing thread, and StateConflictError when optimistic concurrency checks fail during state updates.

ThreadAlreadyExistsError: Preventing Duplicate Thread Creation

Definition and Location

ThreadAlreadyExistsError is defined in [aisuite/agents/runner.py](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/runner.py) (lines 30-44). This exception enforces the rule that once a thread is persisted to a StateStore, it cannot be reinitialized with the same thread_id.

When It Triggers

The check occurs inside the Runner._run_impl method. When you call Runner.run() with both a state_store and thread_id parameter, the code verifies whether state already exists for that identifier:

if state_store is not None and state_store.load_state(thread_id) is not None:
    raise ThreadAlreadyExistsError(...)

If the store returns a non-None state object, the error immediately halts execution. This forces you to use Runner.continue_sync() or Runner.continue() instead, ensuring the original conversation history is preserved and never accidentally overwritten by a fresh run.

StateConflictError: Optimistic Concurrency Control

Definition and Location

StateConflictError is defined in [aisuite/agents/state_store.py](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py) (lines 15-19). This exception prevents lost updates when multiple processes or workers attempt to write to the same thread simultaneously.

Revision Checking Mechanism

The error originates in the _assert_revision helper function (lines 71-80), which validates that the expected revision matches the current stored revision before save_state completes. When StateStore.save_state is invoked with a revision parameter that does not match the stored value, the helper raises StateConflictError:

_assert_revision(thread_id, current.revision if current else None, revision)

This aborts the write operation, allowing the caller to load the latest state and retry with the correct revision number. Both InMemoryStateStore and FileStateStore implement this check to protect against race conditions in multi-worker deployments.

Architectural Flow: How the Errors Work Together

The thread protection mechanism operates across three distinct phases:

1. Thread Initialization – When Runner.run() is called with a thread_id, Runner._run_impl queries the store. If load_state returns existing data, ThreadAlreadyExistsError raises immediately, preventing duplicate creation.

2. State Persistence – After a run completes, Runner._run_impl calls state_store.save_state. The concrete store implementations execute _assert_revision to verify the writer has the latest version. If another process modified the thread since this client loaded it, StateConflictError raises.

3. Thread Continuation – When Runner.continue_* is used, the existing StoredRunState is loaded with its current revision. After appending new messages, the final state saves with the previous revision passed as the revision parameter, enabling the optimistic concurrency check to detect any intermediate modifications.

Code Examples

Handling ThreadAlreadyExistsError

Catch this exception to switch from creating a new thread to continuing an existing one:

import aisuite as ai
from aisuite.agents import Runner

try:
    result = Runner.run(
        agent=my_agent,
        input="Start a new conversation",
        state_store=ai.InMemoryStateStore(),
        thread_id="demo-thread",
    )
except ai.ThreadAlreadyExistsError as exc:
    # The thread already exists – continue instead

    result = Runner.continue_sync(
        target=my_agent,
        input="Add another question",
        state_store=ai.InMemoryStateStore(),
        thread_id="demo-thread",
    )
print(result.final_output)

The except block catches the error and switches to a continuation path, ensuring the prior state is retained.

Retrying StateConflictError with Back-off

Implement retry logic to handle concurrent writes safely:

import time
from aisuite.agents import Runner, InMemoryStateStore, StateConflictError

store = InMemoryStateStore()
thread_id = "shared-thread"

# First run – creates the thread

Runner.run_sync(agent=my_agent, input="Hello", state_store=store, thread_id=thread_id)

# Simulate two workers trying to write concurrently

def safe_save(new_input):
    while True:
        try:
            # Load current state and its revision

            stored = store.load_state(thread_id)
            rev = stored.revision if stored else None
            # Continue the thread with the expected revision

            return Runner.continue_sync(
                target=my_agent,
                input=new_input,
                state_store=store,
                thread_id=thread_id,
                revision=rev,        # <-- passed to StateStore.save_state

            )
        except StateConflictError:
            # Someone else wrote in the meantime – retry after a short back‑off

            time.sleep(0.1)

result = safe_save("What is the weather?")
print(result.final_output)

The loop retries when StateConflictError occurs, guaranteeing that the final write succeeds with the latest revision.

Summary

• ThreadAlreadyExistsError in aisuite/agents/runner.py prevents accidental overwrites by requiring Runner.continue_* for existing threads rather than Runner.run.
• StateConflictError in aisuite/agents/state_store.py implements optimistic concurrency via the _assert_revision helper, blocking writes that would cause lost updates.
• Together they ensure thread immutability and prevent race conditions in distributed, multi-process workloads.
• Both exceptions are exported publicly in aisuite/__init__.py and aisuite/agents/__init__.py for convenient access as ai.ThreadAlreadyExistsError and ai.StateConflictError.

Frequently Asked Questions

What is the difference between ThreadAlreadyExistsError and StateConflictError?

ThreadAlreadyExistsError prevents creating a new thread with an ID that already exists in the store, while StateConflictError prevents concurrent modifications to an existing thread by validating revision numbers during save operations according to the aisuite source code.

How do I continue an existing conversation in aisuite?

Catch ThreadAlreadyExistsError and use Runner.continue_sync() or Runner.continue() with the same thread_id and state_store parameters. This appends new messages to the existing conversation history rather than replacing it.

Can multiple workers safely write to the same thread?

Yes, but you must implement retry logic around StateConflictError. Load the latest state to obtain the current revision, then retry the continue operation. The _assert_revision check in state_store.py ensures only writers with the latest revision succeed.

Where are these exceptions defined in the source code?

ThreadAlreadyExistsError is defined in [aisuite/agents/runner.py](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/runner.py), while StateConflictError is defined in [aisuite/agents/state_store.py](https://github.com/andrewyng/aisuite/blob/main/aisuite/agents/state_store.py). Both are re-exported in aisuite/__init__.py for easy importing.

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