Session Store
Default JSON storage, hierarchical sessions, identity, and task-local context
Session stores persist chat history and metadata โ swap the default JSON backend or use hierarchical forks without changing your agent code.
from praisonaiagents import Agent
agent = Agent(
name="Assistant",
memory={"session_id": "user-42-chat"},
)
agent.start("Remember I like tea.")
agent.start("What do I like?") # History restored from ~/.praisonai/sessions/The user chats across restarts; the session store persists history under ~/.praisonai/sessions/.
How It Works
Where sessions are stored
The session directory is resolved live from PRAISONAI_HOME, so relocating the store root works even after PraisonAI is imported (PraisonAI PR #4125).
Resolution order (highest priority first):
| Source | Wins when | Result |
|---|---|---|
session_dir= constructor arg | Passed to DefaultSessionStore(...) | That exact directory |
praisonaiagents.session.store.DEFAULT_SESSION_DIR = "..." | Explicitly assigned | Pins the root process-wide (test/embedding hook) |
PRAISONAI_HOME env var | Set (no explicit override above) | $PRAISONAI_HOME/sessions/ โ resolved live |
| (default) | Nothing set | ~/.praisonai/sessions/ |
Set PRAISONAI_HOME before the first turn to redirect storage โ it is honoured live, even after praisonaiagents is imported:
import os
os.environ["PRAISONAI_HOME"] = "/data/praisonai" # honoured live, even after import
from praisonaiagents import Agent
agent = Agent(name="Assistant", memory={"session_id": "user-42-chat"})
agent.start("Remember I like tea.") # saved under /data/praisonai/sessions/On-disk filenames (collision-safe)
Two different session ids never share a file. Each user keeps their own history, even when ids differ only in ., @, :, /, +, or spaces.
from praisonaiagents import Agent
alice = Agent(name="Assistant", memory={"session_id": "alice.smith@example.com"})
other = Agent(name="Assistant", memory={"session_id": "alice_smith_example_com"})
# Separate files โ neither agent sees the other's history.| Session id | File |
|---|---|
user-42-chat (only letters, digits, _, -) | user-42-chat.json โ unchanged, no migration |
alice.smith@example.com (any other character) | alice_smith_example_com~7dcd3a39ad3a8d21.json |
Letters and digits include Unicode (cafรฉ keeps cafรฉ.json). Other ids use <readable[:64]>~<sha256(id)[:16]>.json: the readable prefix replaces unsafe characters with _ and is capped at 64 characters; the 16 hex digest of the full id keeps the mapping injective.
- Legacy migration โ a file written by an older release under the old lossy name is moved to the new path on first read, only when its stored
session_idequals the requested id. A legacy file owned by a different id is never adopted. - Ownership check on read โ if a loaded file’s stored
session_iddiffers from the requested id, it is never returned. The store logs a warning, starts a fresh session, and moves the foreign file to its owner’s path (or sets it aside as<name>.json.corrupt-<ts>when that path is taken). - Spill files from write-failure salvage use the same filename stem.
No code change is needed โ existing sessions with safe ids keep their files. See PraisonAI issue #6024 and PR #6061.
Portable export/import
DefaultSessionStore, SqliteSessionStore, and SqliteTranscriptStore all implement PortableSessionStoreProtocol, so you can back up, migrate, or restore live conversation state as a versioned JSON payload.
from praisonaiagents.session import DefaultSessionStore
store = DefaultSessionStore()
payload = store.export_all() # {"version": 1, "sessions": [...]}
report = store.import_sessions(payload) # ImportReport (imported / skipped)- Portability is protocol-level, not store-level โ any store that implements
PortableSessionStoreProtocolgets the same backup/restore surface. - Import resets live routing fields (
gateway_session_id,agent_id) by default, so restored sessions stay inert until re-bound.
See Gateway Session Portability for the operator-facing backup / migrate / restore story.
Quick Start
- Persist with session_id
python from praisonaiagents import Agent agent = Agent( name="Assistant", memory={"session_id": "user-42-chat"}, ) agent.start("Remember I like tea.") agent.start("What do I like?")Default files live at
~/.praisonai/sessions/{session_id}.jsonfor ids made of letters, digits,_and-โ see On-disk filenames for other ids. - Use the store directly
python from praisonaiagents import Agent from praisonaiagents.session import get_default_session_store store = get_default_session_store() session_id = "user-42-chat" agent = Agent(name="Assistant", memory={"session_id": session_id}) store.add_message(session_id, "assistant", agent.start("Summarise our chat"))
Encryption at rest
Wrap any store โ DefaultSessionStore, SqliteSessionStore, HierarchicalSessionStore, or a custom backend โ with EncryptedSessionStore so message content and metadata are encrypted before they reach disk.
import os
from praisonaiagents.session import SqliteSessionStore, EncryptedSessionStore
store = EncryptedSessionStore(
SqliteSessionStore(db_path="~/.praisonai/sessions.db"),
key=os.environ["PRAISONAI_SESSION_KEY"],
)See Encrypted Session Store for keys, honest limits, and migration.
Core Exports
| Export | Purpose |
|---|---|
DefaultSessionStore | JSON-on-disk default backend |
SqliteSessionStore | Indexed variant of the default store โ stdlib sqlite3 + FTS5 for scalable cross-session recall (Issue #2927). See SqliteSessionStore. |
EncryptedSessionStore | Wrap any session store to encrypt message content and metadata at rest โ see Encrypted Session Store |
SessionEncryptionError | Raised on missing/invalid keys, wrong-key decryption, or search() calls on an encrypted store |
generate_session_key() | Generate a Fernet key suitable for PRAISONAI_SESSION_KEY |
SessionMirrorProtocol | Contract for a local-first remote sink โ see Session Mirror |
SessionMessage, SessionData | Typed message and session payloads |
CompactionCheckpoint | Persisted compaction summary + resume anchor โ see Compaction Checkpoints |
get_default_session_store() | Process-wide store accessor |
SessionStoreProtocol | Implement for Redis, Postgres, S3 โ see Session Protocol |
HierarchicalSessionStore, get_hierarchical_session_store() | Forks, snapshots, parent-child โ see Session Hierarchy |
IdentityResolverProtocol, FileIdentityResolver | Map anonymous โ known user IDs across sessions |
SessionContext, set_session_context(), get_session_context() | Task-local session context for async flows |
DefaultSessionStore constructor
| Parameter | Type | Default | Description |
|---|---|---|---|
mirror | Optional[SessionMirrorProtocol] | None | Optional remote sink for local-first mirroring. See Session Mirror. |
SqliteSessionStore
SqliteSessionStore is a drop-in subclass of DefaultSessionStore that keeps JSON transcripts as the durable record and maintains a stdlib sqlite3 index alongside them. It gives you two indexed hot paths instead of directory scans:
- Cross-session search โ FTS5 index of message content, used by
search()(Issue #2927). - Gateway/agent routing โ
session_routeindex ofgateway_session_idandagent_id, used byget_by_gateway_session()andlist_sessions_by_gateway_agent()(Issue #2956).
Both stay independent of the number of stored sessions, so a long-lived gateway bot with thousands of sessions still routes an inbound message in a single indexed lookup.
SqliteSessionStore opens its underlying FTS5 index through sqlite_connect(..., guard=True), so a malformed index file is detected on open, forensically backed up, and repaired in place instead of crashing the store. See Hardened SQLite Storage โ Integrity Guard.
from praisonaiagents import Agent
from praisonaiagents.session import SqliteSessionStore
store = SqliteSessionStore(db_path="~/.praisonai/sessions.db")
agent = Agent(name="Gateway", session_store=store)from praisonaiagents import Agent
from praisonaiagents.session import SqliteSessionStore
store = SqliteSessionStore(db_path="~/.praisonai/sessions.db")
# Long-lived gateway bot: inbound message arrives tagged with a gateway_session_id.
# Routing back to the right session is an indexed lookup, not a directory scan.
agent = Agent(name="Gateway", session_store=store)
agent.start("Handle inbound message", session_id="chat-42")
# Later โ resolve the local session for an inbound gateway event:
session = store.get_by_gateway_session("gw-abc-123")
recent = store.list_sessions_by_gateway_agent("agent-support", limit=20)Constructor
| Parameter | Type | Default | Description |
|---|---|---|---|
session_dir | Optional[str] | None | Directory for session JSON files (as in DefaultSessionStore). |
db_path | Optional[str] | <session_dir>/sessions_index.db | Path to the SQLite index file. Use ":memory:" for an ephemeral in-process index. Supports ~ expansion. |
**kwargs | โ | โ | Forwarded to DefaultSessionStore(...). |
Behaviour
| Aspect | Detail |
|---|---|
| Backend | Stdlib sqlite3 (lazy-imported). Content index: FTS5 virtual table session_fts(session_id UNINDEXED, content) + session_meta(session_id, updated_at). Routing index: session_route(session_id PRIMARY KEY, gateway_session_id, agent_id) with idx_route_gateway and idx_route_agent. |
| Search query | SELECT session_id FROM session_fts WHERE session_fts MATCH ? ORDER BY bm25(session_fts) LIMIT ? |
| Gateway lookup | SELECT session_id FROM session_route WHERE gateway_session_id = ? (overrides parent’s O(N) directory scan) |
| Agent lookup | SELECT session_id FROM session_route WHERE agent_id = ? LIMIT ? (overrides parent’s O(N) directory scan) |
| Fallback | If FTS5 is unavailable โ plain LIKE table. If the routing query fails or sqlite3 is unavailable โ transparent fallback to DefaultSessionStore.get_by_gateway_session / list_sessions_by_gateway_agent scan. |
| Backfill | Legacy JSON transcripts are indexed once, on first search() or first routing lookup. A session is treated as indexed only if it appears in both session_meta and session_route, so stores upgraded from a prior release re-populate route rows on first read. |
| Indexed content | _flatten concatenates archived_messages (compacted turns) ahead of the active messages, so compacted history stays searchable (Issue #5031). The synthetic [Session Summary] message built by retention="compact" (metadata["compaction"]=True) is skipped โ the raw archived turns it paraphrases are already indexed, so indexing the summary too would double-count and spend the bounded candidate allowance on misses (Issue #5051). |
| Content version | INDEX_CONTENT_VERSION = 3 (v2 = archived turns indexed, Issue #5031; v3 = compaction summaries excluded, Issue #5051). Persisted in the session_index_meta table (content_version key). A stale version forces a one-time FTS rebuild on startup. |
| Write sync | _save_session, add_message, and _modify_session_locked (covers set_chat_history, set_gateway_info, append_compaction_checkpoint, and clear_session) all re-index the retained content and routes. delete_session is the only operation that deindexes a session (removes rows from session_meta, session_fts, and session_route). Route rows are INSERT OR REPLACEd when either gateway_session_id or agent_id is set, and DELETEd (via delete_session) when the session is removed. |
| Query rewrite | Free text is tokenised on alphanumerics into an OR of quoted terms for a safe FTS5 MATCH expression. |
| Candidate fan-out | Over-fetches max(limit * 5, limit) candidates so lineage dedup / automated demotion can still promote the right sessions into the final limit. |
| Corrupt-transcript tolerance | search() skips unreadable candidates โ OSError, JSONDecodeError, and UnicodeDecodeError โ on both the FTS5 and LIKE candidate paths, so one bad transcript on disk cannot abort recall across the remaining valid sessions. See Cross-Session Recall โ Resilience to Corrupt Transcripts. |
| Malformed-turn tolerance | SessionData.from_dict skips any entry in messages or archived_messages that is not a JSON object (null, a string, a number). This covers the SQLite indexed route, the SqliteTranscriptStore indexed route, and the default-store scan fallback โ a single malformed turn on disk no longer crashes get_by_gateway_session or list_sessions_by_gateway_agent. The on-disk transcript and its route index record are left untouched. When one or more dropped active turns precede the compaction anchor, last_compaction.message_index is shifted down by the number of entries dropped before it so get_working_history() still slices the retained tail correctly. See PraisonAI PR #5676 / issue #5672. |
A long-lived gateway bot can reset the active window without losing its route:
from praisonaiagents import Agent
from praisonaiagents.session import SqliteSessionStore
store = SqliteSessionStore(db_path="~/.praisonai/sessions.db")
agent = Agent(name="Gateway", session_store=store)
# Session is bound to a gateway event and handles a conversation
agent.start("Hello", session_id="chat-42")
store.set_gateway_info("chat-42", gateway_session_id="gw-abc", agent_id="support")
# Operator clears the active window to start a fresh chat for the same user
store.clear_session("chat-42")
# The next inbound gateway event still routes to the same local session โ
# the route index survives clear_session (PR #5681).
same = store.get_by_gateway_session("gw-abc") # SessionData for "chat-42"
assert same.session_id == "chat-42"Fallback matrix
| Environment | Behaviour |
|---|---|
sqlite3 + FTS5 present (default) | Bounded FTS5 MATCH lookup for search, indexed SELECT on session_route for gateway/agent routing |
sqlite3 present, FTS5 unavailable | Bounded LIKE lookup for search; routing index still active |
sqlite3 unavailable or index open fails | Transparent fallback to DefaultSessionStore.search and DefaultSessionStore.get_by_gateway_session / list_sessions_by_gateway_agent full-directory scans |
Sizing
- Each row in
session_ftsholds the flattened concatenation of a session’s message content (newline-joined). Large transcripts increase index size roughly linearly. - Use
":memory:"for ephemeral tests; use the default disk path for gateway bots that need durability across restarts.
Bookends, automated demotion, and lineage dedup apply to results from both stores โ see Cross-Session Recall.
Upgrade note: archived-history reindex
See Cross-Session Recall โ Recall Across Compacted History for the caller-facing behaviour.
Compaction Checkpoints
When context compaction runs during a conversation, the store can persist the summary so a later resume replays the compacted working history (summary + retained tail) instead of the full raw transcript. See Compacted Session Resume for the end-to-end agent flow.
from praisonaiagents import CompactionCheckpoint
from praisonaiagents.session import DefaultSessionStore
store = DefaultSessionStore()
store.append_compaction_checkpoint("chat-42", "Earlier: we discussed X, Y, Z.")
history = store.get_working_history("chat-42") # summary + tailStore Methods
| Method | Description |
|---|---|
append_compaction_checkpoint(session_id, summary, *, role="system", tokens_before=0, tokens_after=0, metadata=None) | Persist a checkpoint anchored to the current end of the transcript. Returns bool; a blank/whitespace summary is a no-op returning False. |
get_working_history(session_id, max_messages=None) | Canonical read path โ uses the checkpoint when present (summary + tail), falls back to raw chat history when not. |
SessionData additions
SessionData.last_compaction holds the latest CompactionCheckpoint (or None). Two helpers support cheap resume:
| Member | Description |
|---|---|
last_compaction | Optional[CompactionCheckpoint] โ the persisted checkpoint, serialised into the session JSON |
trim_messages(max_messages) | Trim the transcript head, shifting the checkpoint anchor so the retained tail stays aligned |
get_working_history(max_messages=None) | Reconstruct [summary_message, *tail]; falls back to get_chat_history with no checkpoint |
Task-Local Context
from praisonaiagents.session import set_session_context, get_session_context
set_session_context(session_id="batch-job-1", user_id="operator")
ctx = get_session_context()
print(ctx.session_id)Best Practices
Prefer session_id on Agent over manual store calls
Let Agent(memory={"session_id": "..."}) handle persistence โ use the store directly only for admin, migration, or custom backends.
Use hierarchical store for forks and snapshots
Switch to get_hierarchical_session_store() when you need branching conversations or revert โ see Session Hierarchy.
Set task-local context in async workers
Call set_session_context() at the start of each async task so downstream code reads the correct session without threading IDs through every call.