How to Implement Thread Continuation Across Processes Using aisuite's PostgresStateStore

aisuite's PostgresStateStore persists the full execution state of an agent thread in PostgreSQL, allowing any process to resume a conversation by thread_id while guaranteeing exactly-once semantics through revision-based optimistic locking.

The andrewyng/aisuite repository provides a production-ready state store that keeps messages, step identifiers, and a revision counter durable across restarts. By leveraging the store, you can pause an agent in one Python process and safely continue execution in another worker or after a full application restart.

Initialize PostgresStateStore for Cross-Process Persistence

To begin, instantiate the store with a PostgreSQL connection string. The from_dsn class method in aisuite/agents/postgres_state_store.py creates a live psycopg connection and optionally builds the required schema.

import os
from aisuite import ai

dsn = os.getenv("POSTGRES_DSN")  # e.g. "postgresql://user:pwd@host/db"

store = ai.PostgresStateStore.from_dsn(dsn, create_schema=True)

Passing create_schema=True invokes PostgresStateStore.create_schema, which auto-generates tables such as agent_thread_heads so the store is ready for immediate use.

Persist RunState with Optimistic Revision Locking

After each LLM turn, call save_state with the thread identifier and the previous revision. As implemented in aisuite/agents/postgres_state_store.py, the method executes the following in a single transaction:

1. Loads the current head row via _load_head_for_update.
2. Validates the revision using _assert_revision from aisuite/agents/state_store.py.
3. Inserts new messages into the messages table.
4. Updates agent_thread_heads and returns a new StoredRunState with an incremented revision.

from aisuite.framework import RunState, Message

thread_id = "thread-1234"
state = RunState(messages=[])

new_msg = Message(role="assistant", content="Hello from a new process!")
state.messages.append(new_msg)

# First save: revision=None signals a new thread

stored = store.save_state(thread_id, state, revision=None)
print(f"Saved revision {stored.revision} at {stored.updated_at}")

You must retain the returned revision to safely append subsequent steps and prevent concurrent overwrites.

Resume Thread Continuation in a Separate Process

Any worker that knows the thread_id can resume execution. The load_state method reads the head row, fetches associated messages via _load_messages, and reconstructs the RunState defined in aisuite/framework/run_state.py.

In a second script or worker:

from aisuite import ai

store = ai.PostgresStateStore.from_dsn(dsn)
saved = store.load_state("thread-1234")

if saved:
    current_state = saved.state
    print(f"Loaded {len(current_state.messages)} prior messages")

    # Append the next LLM turn

    next_msg = Message(role="assistant", content="Continuing from another process!")
    current_state.messages.append(next_msg)

    # Enforce sequential consistency with the stored revision

    new_stored = store.save_state(
        "thread-1234", current_state, revision=saved.revision
    )
    print(f"Advanced to revision {new_stored.revision}")

Because the underlying data lives in PostgreSQL rather than in-memory, the second process receives the exact conversation state. The Message dataclass from aisuite/framework/message.py ensures that roles and content are reconstructed identically.

Manage Long Contexts with Message Compaction

For long-running threads, the store separates the LLM window from full history:

• model_context_message_ids — The ordered list of message IDs presented to the LLM.
• full_history_message_ids — The complete log of every message in the conversation.

These columns live in the agent_thread_heads table. When a thread exceeds context limits, you can call compact_state to summarize older messages. The list_compactions method lets you inspect compaction records later, keeping the active context window small without losing the full audit trail.

Summary

• PostgresStateStore in aisuite/agents/postgres_state_store.py makes agent threads durable by persisting state to PostgreSQL.
• from_dsn with create_schema=True bootstraps the schema automatically.
• save_state uses _load_head_for_update and _assert_revision in aisuite/agents/state_store.py to provide optimistic locking and exactly-once write semantics.
• load_state reconstructs the full RunState in any process, enabling seamless thread continuation across workers.
• model_context_message_ids and full_history_message_ids let you optimize LLM windows independently of permanent conversation logs.

Frequently Asked Questions

How does aisuite prevent two processes from overwriting the same thread?

PostgresStateStore uses optimistic locking. Every save_state call must supply the revision returned by the previous save. The _assert_revision helper in aisuite/agents/state_store.py raises a StateConflictError if the provided revision does not match the current database row. This serializes writers so that only one update succeeds per step.

Can multiple workers read the same thread simultaneously?

Yes. load_state issues read-only queries against PostgreSQL, so any number of workers can inspect a conversation concurrently. Conflicts are only possible during writes, which are guarded by the revision check in save_state.

What identifier should I use for a thread ID?

A thread_id is any unique string that you generate, such as a UUID or a user-session key. The store scopes all state in the agent_thread_heads table to this string, so every process that passes the same ID sees the same message history and revision.

Does PostgresStateStore require manual table creation?

No. Passing create_schema=True to from_dsn invokes PostgresStateStore.create_schema, which creates the necessary tables and indices automatically. You only need to enable this during the first initialization or when migrating schemas.

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