Skip to main content
Advanced session management with forking, snapshots, and revert capabilities for complex conversation flows.

​
Quick Start

1

Agent-Centric Usage

Use HierarchicalSessionStore automatically when fork/snapshot operations are needed:
from praisonaiagents import Agent
from praisonaiagents.session import get_hierarchical_session_store

# Agent with hierarchical session support
agent = Agent(
    name="Assistant",
    memory={"session_id": "project-alpha"},
)

response = agent.start("Let's plan the migration")

# Later - fork the conversation to explore alternatives
store = get_hierarchical_session_store()
fork_id = store.fork_session("project-alpha")
2

Direct Store Usage

For advanced control, use the store directly:
from praisonaiagents.session import HierarchicalSessionStore

# Create a hierarchical session store
store = HierarchicalSessionStore(session_dir="./sessions")

# Create a session
session_id = store.create_session(title="My Project")

# Add messages
store.add_message(session_id, "user", "Hello!")
store.add_message(session_id, "assistant", "Hi there!")

# Create a snapshot
snapshot_id = store.create_snapshot(session_id, label="Checkpoint 1")

# Continue working...
store.add_message(session_id, "user", "Do something risky")

# Revert to snapshot if needed
store.revert_to_snapshot(session_id, snapshot_id)

​
Multi-Process Safety

HierarchicalSessionStore is safe under concurrent multi-process and multi-instance use, with the same guarantees as DefaultSessionStore plus extended-field preservation:
  • File locking — fcntl.flock() on Unix, msvcrt.locking() on Windows
  • Atomic writes — temp file + os.replace() prevents partial-write corruption
  • Reload under lock — every mutator (add_message, update_session_metadata, clear_session, set_agent_info, set_gateway_info) reloads the session from disk inside the FileLock before mutating, so two processes sharing the same session directory cannot drop each other’s messages.
  • Extended-field preservation — parent_id, children_ids, snapshots, and forked_from_message_id survive across update_session_metadata and clear_session calls (fixed in PR #1745). Earlier releases could silently wipe these fields on the next save after a metadata update.
This matters most for the PraisonAI UI host, where the UI calls add_message from the request thread while the agent’s auto_save writes assistant turns from a worker thread, and _persist_session_stats() runs update_session_metadata after every turn. PR #1745 closes the last remaining race in that path.

​
Configuration Options

HierarchicalSessionStore API Reference

Complete API documentation with all parameters and return types

​
Extended fields preserved across mutators

FieldSet byPreserved by
parent_idcreate_session(parent_id=...), fork_session(...)all mutators (PR #1745)
children_idscreate_session(parent_id=...), fork_session(...)all mutators (PR #1745)
snapshotscreate_snapshot(...)all mutators (PR #1745)
forked_from_message_idfork_session(from_message_index=...)all mutators (PR #1745)

​
Key Methods

class HierarchicalSessionStore(DefaultSessionStore):
    def create_session(
        self,
        session_id: Optional[str] = None,
        title: Optional[str] = None,
        parent_id: Optional[str] = None,
        agent_name: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> str:
        """Create a new session, optionally as child of another."""
    
    def fork_session(
        self,
        session_id: str,
        from_message_index: Optional[int] = None,
        title: Optional[str] = None
    ) -> str:
        """Fork a session from a specific message point."""
    
    def create_snapshot(
        self,
        session_id: str,
        label: Optional[str] = None
    ) -> str:
        """Create a labeled snapshot of the current session state."""
    
    def revert_to_snapshot(
        self,
        session_id: str,
        snapshot_id: str
    ) -> bool:
        """Revert session to a previous snapshot."""
    
    def get_snapshots(self, session_id: str) -> List[SessionSnapshot]:
        """Get all snapshots for a session."""
    
    def export_session(self, session_id: str) -> Dict[str, Any]:
        """Export session data for transfer."""
    
    def import_session(self, data: Dict[str, Any]) -> str:
        """Import a session from exported data."""

​
Common Patterns

​
Forking Sessions

store = HierarchicalSessionStore()

# Create main session
main_id = store.create_session(title="Main Branch")
store.add_message(main_id, "user", "Start project")
store.add_message(main_id, "assistant", "Project initialized")

# Fork from message index 1
fork_id = store.fork_session(
    main_id,
    from_message_index=1,
    title="Experimental Branch"
)

# Fork has messages up to index 1
# Can now diverge independently
store.add_message(fork_id, "user", "Try experimental approach")

​
Snapshot Management

store = HierarchicalSessionStore()
session_id = store.create_session()

# Work and create snapshots
store.add_message(session_id, "user", "Phase 1")
snap1 = store.create_snapshot(session_id, label="After Phase 1")

store.add_message(session_id, "user", "Phase 2")
snap2 = store.create_snapshot(session_id, label="After Phase 2")

# List all snapshots
snapshots = store.get_snapshots(session_id)
for snap in snapshots:
    print(f"{snap.label}: {snap.message_count} messages")

# Revert to Phase 1
store.revert_to_snapshot(session_id, snap1)

​
Export/Import

# Export from one store
store1 = HierarchicalSessionStore(session_dir="./store1")
session_id = store1.create_session(title="Portable Session")
store1.add_message(session_id, "user", "Important data")

exported = store1.export_session(session_id)

# Import to another store
store2 = HierarchicalSessionStore(session_dir="./store2")
new_id = store2.import_session(exported)

​
Best Practices

Provide descriptive titles to help identify sessions later:
store.create_session(title="Project Alpha - Requirements Phase")
store.fork_session(session_id, title="Alternative Approach")
Take snapshots before potentially destructive changes:
# Before making experimental changes
snap_id = store.create_snapshot(session_id, label="Before refactor")
# ... make changes ...
# Revert if needed
store.revert_to_snapshot(session_id, snap_id)
Regularly remove unused sessions to save disk space:
# List all sessions
sessions = store.list_sessions()
for session in sessions:
    if should_delete(session):
        store.delete_session(session.session_id)
Fork sessions to explore different conversation paths:
# Main conversation path
main_id = store.create_session(title="Main Discussion")
store.add_message(main_id, "user", "Let's solve this problem")

# Experimental path
experiment_id = store.fork_session(main_id, title="Experimental Solution")
store.add_message(experiment_id, "user", "What if we try a different approach?")

Session Persistence

Basic session persistence with zero configuration

Session Protocol

Custom session store implementation guide