SQLiteSessionStorage#

Restore model context and preserve immutable original messages. See the guide.

class zett_agent.storage.SQLiteSessionStorage(path=None)[source]#

Standalone SQLite session storage shipped with zett-agent.

The first append creates session identity and history. Root sessions store no parent; delegated sessions store a plain parent ID without a foreign key. Message metadata and tags live in each Raw Log message JSON envelope. Only compaction creates snapshots.

Every storage operation is asynchronous and runs on SQLAlchemy’s asyncio SQLite driver, so persistence never blocks the Agent event loop. The file path is prepared during construction; ORM tables are created by the first awaited call, which keeps construction usable before a loop exists.

Parameters:: path (str | Path | None) – SQLite file path. None uses ~/.zett-agent/sessions.sqlite3. Parent directories are created during construction.

Note

Close the owned connection pool with await storage.close(). Tests must pass a temporary file and never use the default user database.

Examples

Use a temporary database in tests:

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as directory:
    storage = SQLiteSessionStorage(Path(directory) / "sessions.sqlite3")
    try:
        session = await storage.create_session(title="Documentation review")
        assert await storage.get_session(session.session_id) == session
    finally:
        await storage.close()

        flowchart LR
    raw[Raw Log: immutable UI history]
    snapshot[Snapshot: compacted context + boundary]
    tail[Raw Log after boundary]
    view[SessionView]
    snapshot --> view
    raw --> tail
    tail --> view

Restoring one session view#

Warning

The default path writes personal data. Libraries and tests should pass an explicit path instead of silently using the default location.

See also

SQLiteSessionExtension integrates this storage with lifecycle hooks, and Implement a storage adapter explains the generic storage boundary.

async close()[source]#

Release the connection pool without deleting stored data.

Return type:: None

async create_session(*, session_id=None, parent_session_id=None, title=None, agent_name=None, session_type=0)[source]#

Create an empty session before its first Raw Log message is appended.

Parameters:

session_id (str | None)
parent_session_id (str | None)
title (str | None)
agent_name (str | None)
session_type (int)

Return type:

SessionSummary

async list_raw_messages(session_id, *, after_sequence=0, through_sequence=None, limit=10000, offset=0)[source]#

Return one chronological page from a session’s immutable Raw Log.

This query deliberately ignores snapshots: snapshots are compact model context, while a conversation UI must display the original messages. after_sequence and through_sequence optionally restrict the Raw Log sequence range; offset and limit paginate that filtered range.

Parameters:

session_id (str) – Conversation to query.
after_sequence (int) – Exclusive lower sequence bound, defaulting to zero.
through_sequence (int | None) – Inclusive upper bound, or None for no upper bound.
limit (int) – Positive page size.
offset (int) – Non-negative number of matching messages to skip.

Returns:

Typed raw records in ascending sequence order; empty if none match.

Raises:

ValueError – If a bound or pagination argument is outside its allowed range.

Return type:

list[RawMessageRecord]

async list_sessions(*, session_types=None, limit=100, offset=0)[source]#

Return one page of root and subagent sessions by latest activity.

Parameters:

session_types (Sequence[int] | None)
limit (int)
offset (int)

Return type:

list[SessionSummary]

async count_messages(session_id)[source]#

Count immutable raw messages without reconstructing their payloads.

Parameters:: session_id (str)
Return type:: int

async get_session(session_id)[source]#

Return one session’s storage metadata without loading its messages.

Parameters:: session_id (str)
Return type:: SessionSummary | None

async update_session(session_id, *, title=None, agent_name=None)[source]#

Update mutable session fields and return the refreshed typed summary.

Session identity, parent linkage, and Raw Log messages are immutable. Display values are normalized before persistence. None leaves that field unchanged; at least one field must be supplied.

Parameters:

session_id (str)
title (str | None)
agent_name (str | None)

Return type:

SessionSummary | None

async delete_session(session_id)[source]#

Explicitly delete raw messages and snapshots owned by one session.

Parameters:: session_id (str)
Return type:: bool

async load(session_id)[source]#

Restore the latest checkpoint and the original messages after its boundary.

Parameters:: session_id (str) – Stable conversation identifier.
Returns:: A typed SessionView containing the snapshot and ordered raw tail. An unknown session produces an empty view without creating records.
Return type:: SessionView

async append(session_id, request_id, message, timing=None, parent_session_id=None, title=None, agent_name=None, metadata=None, tags=None, usage=None)[source]#

Append one original message and return its new per-session sequence.

Parameters:

session_id (str) – Owning conversation, created on first append if needed.
request_id (str) – Correlation ID; repeated IDs do not deduplicate writes.
message (AnyMessage) – Original provider-neutral user, assistant, tool, or agent input.
timing (MessageTiming | None) – Observed operation timing; None records an instantaneous append.
parent_session_id (str | None) – Immutable parent link for delegated conversations.
title (str | None) – Optional display title used when creating the session.
agent_name (str | None) – Optional display name used when creating the session.
metadata (Mapping[str, JsonValue] | None) – Request-scoped JSON data, separate from model message content.
tags (Mapping[str, JsonValue] | None) – JSON classifications with non-empty keys.
usage (ModelUsage | None) – Normalized provider counters, allowed only for assistant messages.

Returns:

The strictly increasing sequence assigned to the stored record.

Raises:

ValueError – If metadata, session display values, parent linkage, or the message/usage combination is invalid.

Return type:

int

async snapshot(session_id, compacted_message, compacted_through_sequence, expected_version)[source]#

Create an immutable compaction checkpoint using optimistic version checking.

Parameters:

session_id (str) – Conversation whose context was compacted.
compacted_message (CompactedMessage) – Summary replacing older dialogue in model context.
compacted_through_sequence (int) – Last Raw Log message represented by the summary; later records remain the replay tail.
expected_version (int) – Last snapshot version seen by the caller, or zero before the first checkpoint.

Returns:

The new checkpoint with its incremented version and UTC timestamps.

Raises:

ValueError – If the version is stale, the boundary is outside the Raw Log, or the boundary does not advance beyond the previous snapshot.

Return type:

ContextSnapshot

Note

This operation never changes or deletes original Raw Log messages.

sync(*, runtime=None)#

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

Examples

Use a model or tool from a plain function:

with read_file.sync() as read:
    result = read({"path": "README.md"})

Parameters:: runtime (SyncRuntime | None)
Return type:: SyncObject[Self]