haive.core.engine.agent.persistence.manager

PostgreSQL persistence manager for the Haive framework.

This module provides a comprehensive persistence manager that integrates Supabase authentication with PostgreSQL persistence for agent state management. It centralizes thread registration, checkpoint management, and connection pool handling in a robust and reusable design.

The PersistenceManager class serves as the primary integration point between the HaiveRunnableConfigManager and the underlying PostgreSQL database.

Classes

PersistenceManager

Manages state persistence for agents, abstracting the complexity of different.

Module Contents

class haive.core.engine.agent.persistence.manager.PersistenceManager(config=None)

Manages state persistence for agents, abstracting the complexity of different. checkpointer implementations and integrating with Supabase authentication.

This manager handles: 1. Auto-detection of available persistence options 2. Configuration of checkpointers (PostgreSQL, Memory) 3. Setup of database connections and pools 4. Thread registration with user context from Supabase 5. Integration with HaiveRunnableConfigManager for authentication

Initialize persistence manager with optional configuration.

Parameters:

config (dict[str, Any] | None) – Optional configuration for persistence

close_pool_if_needed()

Close the PostgreSQL connection pool if it was opened by this manager.

Return type:

None

create_runnable_config(thread_id=None, user_info=None, **kwargs)

Create a RunnableConfig with proper thread ID and authentication context.

Parameters:

• thread_id (str | None) – Optional thread ID for persistence

• user_info – Optional user information dictionary (Supabase)

• **kwargs – Additional runtime configuration

Returns:

RunnableConfig with thread ID and authentication context

delete_thread(thread_id)

Delete a thread from the PostgreSQL database.

Parameters:

thread_id (str) – Thread ID to delete

Returns:

True if deletion succeeded, False otherwise

ensure_pool_open()

Ensure the PostgreSQL connection pool is open.

Returns:

True if the pool was opened or is already open, False otherwise

Return type:

bool

classmethod from_config(db_host='localhost', db_port=5432, db_name='postgres', db_user='postgres', db_pass='postgres', use_async=False, use_pool=True, setup_needed=True)

Create a PersistenceManager from database configuration.

Parameters:

• db_host – Database host

• db_port – Database port

• db_name – Database name

• db_user – Database user

• db_pass – Database password

• use_async – Whether to use async connections

• use_pool – Whether to use connection pooling

• setup_needed – Whether table setup is needed

Returns:

Configured PersistenceManager

classmethod from_env()

Create a PersistenceManager from environment variables.

Returns:

Configured PersistenceManager

Return type:

Any

get_checkpointer(persistence_type=None, persistence_config=None)

Create and return the appropriate checkpointer based on configuration and available dependencies.

Parameters:

• persistence_type – Optional persistence type override

• persistence_config (dict[str, Any] | None) – Optional persistence configuration override

Returns:

A configured checkpointer instance

static get_or_create_thread_id(config=None)

Get thread ID from config or create a new one.

Parameters:

config (dict[str, Any] | None) – Optional RunnableConfig

Returns:

Thread ID string

list_threads(user_id=None, thread_id=None, limit=100, offset=0)

List threads from the PostgreSQL database.

Parameters:

• user_id (str | None) – Optional user ID filter

• thread_id (str | None) – Optional thread ID filter for single thread lookup

• limit (int) – Maximum number of threads to return

• offset (int) – Offset for pagination

Returns:

List of thread information dictionaries

prepare_for_agent_run(thread_id=None, user_info=None, **kwargs)

Comprehensive preparation for an agent run, handling thread registration,. configuration creation, and database setup.

Parameters:

• thread_id (str | None) – Optional thread ID for persistence

• user_info – Optional user information dictionary (Supabase)

• **kwargs – Additional runtime configuration

Returns:

Tuple of (RunnableConfig, current_thread_id)

register_thread(thread_id, auth_info=None)

Register a thread in the PostgreSQL database, including user context from Supabase.

Parameters:

• thread_id (str) – Thread ID to register

• auth_info – Optional authentication information

Returns:

True if registration succeeded, False otherwise

setup()

Setup the checkpointer, including database tables if needed.

Returns:

True if setup succeeded, False otherwise

Return type:

bool