实现存储适配器#

把生命周期集成留在基类扩展中#

不要在新的数据库扩展里重复实现上下文恢复、Raw Log 事件或压缩的记账逻辑。请实现 SessionStorage 契约，并把它传给 BaseSessionPersistenceExtension。基类已经知道何时加载、追加、生成快照，以及清理请求级的记账数据。

运行时的三个存储操作#

SessionStorage 的边界#
方法	返回值	不变量
load	SessionView：可选的检查点，以及有序的原始消息尾部	一致的视图；未知会话为空。
append	分配的每会话序列号	原始消息不可变；追加会原子地推进序列号。
snapshot	带新版本的 ContextSnapshot	检查 expected_version 以及不断推进的 Raw Log 边界；绝不改写原始消息。

会话标题的 CRUD 以及分页的界面历史记录由具体的存储实现提供，例如 SQLiteSessionStorage。它与运行时持久化扩展要求的三个操作相互独立。

完整的适配器示例#

这个适配器在 SQLite 之上加了审计追踪。它有意把实际事务交给经过测试的 SQLite 实现去处理；这并不表示它实现了一个不同的数据库。PostgreSQL 或远程适配器会替换这些实现体，同时保持相同的顺序/版本保证。

class AuditedStorage:
    """A transparent adapter with an audit trail around an existing SQLite store."""

    def __init__(self, database: SQLiteSessionStorage) -> None:
        self.database = database
        self.operations: list[str] = []

    async def load(self, session_id: str) -> SessionView:
        self.operations.append("load")
        return await self.database.load(session_id)

    async def append(
        self,
        session_id: str,
        request_id: str,
        message: AnyMessage,
        timing: MessageTiming | None = None,
        parent_session_id: str | None = None,
        title: str | None = None,
        agent_name: str | None = None,
        metadata: Mapping[str, JsonValue] | None = None,
        tags: Mapping[str, JsonValue] | None = None,
        usage: ModelUsage | None = None,
    ) -> int:
        self.operations.append(f"append:{message.role.value}")
        return await self.database.append(
            session_id, request_id, message, timing, parent_session_id, title, agent_name, metadata, tags, usage
        )

    async def snapshot(
        self,
        session_id: str,
        compacted_message: CompactedMessage,
        compacted_through_sequence: int,
        expected_version: int,
    ) -> ContextSnapshot:
        self.operations.append("snapshot")
        return await self.database.snapshot(session_id, compacted_message, compacted_through_sequence, expected_version)

扩展本身保持精简：

class AuditedPersistence(BaseSessionPersistenceExtension[AuditedStorage]):
    """Only construct the adapter; the base owns all lifecycle integration."""

    def __init__(self, database: SQLiteSessionStorage) -> None:
        super().__init__(AuditedStorage(database))

运行完整程序来验证一次请求会产生 load, append:system, append:user, append:assistant，并且不会做不必要的快照。

快照一致性检查清单#

要一致地读取检查点及其尾部消息；不要混合不同的数据库版本。
把 compacted_through_sequence 视为包含端点的已表示边界。
重放的尾部消息严格从该边界之后开始。
过期的 expected_version 应当被拒绝，而不是静默覆盖检查点。
不要每成功得到一次响应就存一个检查点。
要保留消息角色、属性、重放块、时间信息、元数据、标签和用量。
不要把 request_id 当作自动的幂等键。

为空会话、两消息轮次、工具往返、压缩之后又有新尾部、冲突、未知会话以及数据库重新打开等情况分别编写独立的测试。使用临时数据库，并释放所有连接池。

配套的检查点类型#

快照契约使用来自其定义模块的这个类：

from zett_agent.extensions.compaction import CompactedMessage

checkpoint = CompactedMessage(content="The user prefers Python.")

CompactedMessage 是一种特殊的系统消息，用来标记上下文检查点。它的 content保存摘要；继承来的 attributes保存可选的消息注解。它不会从包的根模块导出。请把这个检查点与原始 Raw Log 分开存储，并连同上面描述的、包含端点的序列边界一起保存。