Sessions

Persist conversation history across agent runs using pluggable storage backends.

Session management is experimental. APIs may change in future releases.

Sessions let your agent remember previous conversations. When you attach a SessionManager to an agent, it automatically loads history at the start of each run and saves it when the run completes — so the agent picks up exactly where the last conversation left off.

Quick start

1. Initialize a storage engine

   Create a SQLite engine and set up the database tables. Do this once at application startup.

   from nexau.archs.session.orm import SQLDatabaseEngine
   
   engine = SQLDatabaseEngine.from_url("sqlite+aiosqlite:///sessions.db")
   await engine.setup_models()
   
2. Create a SessionManager

   from nexau.archs.session import SessionManager
   
   session_manager = SessionManager(engine=engine)
   
3. Attach the manager to your agent

   Pass session_manager, user_id, and session_id when constructing the agent.

   from nexau.archs.main_sub.agent import Agent, AgentConfig
   
   agent = Agent(
       config=agent_config,
       session_manager=session_manager,
       user_id="user_123",
       session_id="sess_456",
   )
   
   response = await agent.run_async("Hello, how are you?")
   # The next run loads this conversation automatically
   response2 = await agent.run_async("What did I just ask?")
   

If you omit session_id, a new session is created. Pass the returned session ID in subsequent requests to continue the same conversation.

Storage backends

Choose a backend based on your deployment needs.

SQLite

File-based storage. No extra dependencies — good for development and small single-process deployments.

from nexau.archs.session.orm import SQLDatabaseEngine
from nexau.archs.session import SessionManager

engine = SQLDatabaseEngine.from_url("sqlite+aiosqlite:///sessions.db")
await engine.setup_models()

session_manager = SessionManager(engine=engine)

The database is a single file you can copy or back up directly.

PostgreSQL / MySQL

SQL databases with full concurrent access. Use these in production.

from nexau.archs.session.orm import SQLDatabaseEngine
from nexau.archs.session import SessionManager

# PostgreSQL
engine = SQLDatabaseEngine.from_url(
    "postgresql+asyncpg://user:pass@localhost/db"
)

# MySQL
engine = SQLDatabaseEngine.from_url(
    "mysql+aiomysql://user:pass@localhost/db"
)

await engine.setup_models()
session_manager = SessionManager(engine=engine)

Install the required driver for your database:

# PostgreSQL
uv pip install asyncpg

# MySQL
uv pip install aiomysql

In-memory

Stores sessions in process memory. Data is lost when the process exits.

from nexau.archs.session.orm import InMemoryDatabaseEngine
from nexau.archs.session import SessionManager

engine = InMemoryDatabaseEngine()
session_manager = SessionManager(engine=engine)

Use in-memory storage for testing only. All session data is lost on process exit.

JSONL

Stores sessions as line-delimited JSON files. Human-readable and easy to inspect or back up.

from nexau.archs.session.orm import JSONLDatabaseEngine
from nexau.archs.session import SessionManager

engine = JSONLDatabaseEngine(storage_path="/path/to/storage")
session_manager = SessionManager(engine=engine)

No database dependency required.

Concurrency control

When multiple requests try to run against the same session simultaneously, SessionManager uses a distributed lock to prevent conflicting writes. Only one agent run can modify a session at a time.

session_manager = SessionManager(
    engine=engine,
    lock_ttl=30.0,           # Lock expires after 30 seconds
    heartbeat_interval=10.0, # Renew lock every 10 seconds
)

Parameter	Default	Description
lock_ttl	30.0	Seconds before a lock expires automatically. Prevents deadlocks if a run crashes.
heartbeat_interval	10.0	Seconds between lock renewals during a long-running operation.

For operations that take longer than the default TTL, increase lock_ttl and set heartbeat_interval to roughly one-third of it.

Common errors

TableNotFoundError: Table 'sessions' does not exist

You haven’t called setup_models() before using the engine. Add it to your application startup:

engine = SQLDatabaseEngine.from_url("sqlite+aiosqlite:///sessions.db")
await engine.setup_models()  # Required before first use

Conversation history not persisting between runs

The agent was created without a session_manager, or without a session_id to resume. Both are required to continue an existing session:

# Wrong — no session manager
agent = Agent(config=agent_config)

# Correct — session manager and IDs provided
agent = Agent(
    config=agent_config,
    session_manager=session_manager,
    user_id="user_123",
    session_id="sess_456",
)

AgentLockExpiredError: Lock expired

The agent run took longer than lock_ttl. Increase the TTL and the heartbeat interval proportionally:

session_manager = SessionManager(
    engine=engine,
    lock_ttl=60.0,
    heartbeat_interval=20.0,
)