SQLiteSessionStorage

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

class zett_agent.storage.SQLiteSessionStorage(path=None)

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.

参数:

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

备注

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

示例

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

警告

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

参见

SQLiteSessionExtension integrates this storage with lifecycle hooks, and 实现存储适配器 explains the generic storage boundary.

async close()

Release the connection pool without deleting stored data.

返回类型:

None

async create_session(*, session_id=None, parent_session_id=None, title=None, agent_name=None, session_type=0)

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

参数:

• session_id (str | None)

• parent_session_id (str | None)

• title (str | None)

• agent_name (str | None)

• session_type (int)

返回类型:

SessionSummary

async list_raw_messages(session_id, *, after_sequence=0, through_sequence=None, limit=10000, offset=0)

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.

参数:

• 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.

返回:

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

抛出:

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

返回类型:

list[RawMessageRecord]

async list_sessions(*, session_types=None, limit=100, offset=0)

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

参数:

• session_types (Sequence[int] | None)

• limit (int)

• offset (int)

返回类型:

list[SessionSummary]

async count_messages(session_id)

Count immutable raw messages without reconstructing their payloads.

参数:

session_id (str)

返回类型:

int

async get_session(session_id)

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

参数:

session_id (str)

返回类型:

SessionSummary | None

async update_session(session_id, *, title=None, agent_name=None)

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.

参数:

• session_id (str)

• title (str | None)

• agent_name (str | None)

返回类型:

SessionSummary | None

async delete_session(session_id)

Explicitly delete raw messages and snapshots owned by one session.

参数:

session_id (str)

返回类型:

bool

async load(session_id)

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

参数:

session_id (str) -- Stable conversation identifier.

返回:

A typed SessionView containing the snapshot and ordered raw tail. An unknown session produces an empty view without creating records.

返回类型:

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)

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

参数:

• 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.

返回:

The strictly increasing sequence assigned to the stored record.

抛出:

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

返回类型:

int

async snapshot(session_id, compacted_message, compacted_through_sequence, expected_version)

Create an immutable compaction checkpoint using optimistic version checking.

参数:

• 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.

返回:

The new checkpoint with its incremented version and UTC timestamps.

抛出:

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

返回类型:

ContextSnapshot

备注

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.

示例

Use a model or tool from a plain function:

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

参数:

runtime (SyncRuntime | None)

返回类型:

SyncObject[Self]