数据持久化 API

持久化适配器用特定数据库支持 Flue 的持久状态。本页记录适配器契约：PersistenceAdapter 接口、connect() 返回的三个存储接口、每个后端必须遵守的存储规则，以及验证实现的契约测试套件。类型和辅助函数从 @flue/runtime/adapter 导出；测试套件从 @flue/runtime/test-utils 导出。有关在项目中配置数据库（db.ts、内存默认值、生态系统适配器），参见数据库；有关持久执行引擎如何使用此存储，参见持久性。

每个后端只有一个适配器契约——没有仅 SQL 或”专家”层级。方法不变量以可观察行为而非存储原语声明，因此非 SQL 后端是一等实现。导出的类型带有详尽的逐方法文档注释，契约测试套件是可执行规范；当所有三个套件通过时，适配器就是正确的。如果本页和包有差异，以包为准。

适配器表面仅限 Node：在 Cloudflare 目标上，每个智能体实例在其 Durable Object 的内置 SQLite 存储中持久化，db.ts 文件在构建时被拒绝，自定义适配器不适用。稳定性：AgentSubmissionStore 的结算和租约方法组镜像持久执行引擎，在 1.0 之前可能会变化，对所有后端均等。

PersistenceAdapter

interface PersistenceAdapter {
  connect(): PersistenceStores | Promise<PersistenceStores>;
  migrate?(): void | Promise<void>;
  close?(): void | Promise<void>;
}

适配器包导出返回此接口的工厂函数；用户从 db.ts 默认导出结果。内置参考实现是 @flue/runtime/node 中的 sqlite(path?: string)。

• connect() — 打开数据库并返回每个存储。启动时等待一次，因此异步池设置、远程握手以及——对于没有 migrate 的适配器——格式版本检查属于这里。不可达的数据库在启动时失败，而不是在第一个请求中。
• migrate() — 将存储带到当前格式版本。启动时在 connect() 之前调用一次。创建任何缺失的模式，首次创建存储时持久记录格式版本，当存储记录了未知或更新版本时大声失败。隐式创建模式的适配器可以省略它，但必须在创建存储的路径中遵守版本义务。
• close() — 释放资源（连接池、文件句柄）。在关闭时调用。

PersistenceStores

interface PersistenceStores {
  readonly submissionStore: AgentSubmissionStore;
  readonly conversationStreamStore: ConversationStreamStore;
  readonly attachmentStore: AttachmentStore;
}

• submissionStore — 持久提交生命周期存储。
• conversationStreamStore — 按智能体实例的规范对话流。
• attachmentStore — 规范对话记录引用的不可变附件字节。

AgentSubmissionStore

持久提交账本：已接纳的载荷、队列排序、尝试/租约协调和结算投影。提交状态遵循 queued → running → (terminalizing →) settled，带有一对 joining/joined 用于在轮次边界吸收进另一个提交实时响应的排队投递。会话在智能体实例生命周期内是追加式的；契约没有按会话删除。每个方法的完整不变量在导出类型的文档注释中。

interface AgentSubmissionStore {
  // 查询
  getSubmission(submissionId: string): Promise<AgentSubmission | null>;
  hasUnsettledSubmissions(): Promise<boolean>;
  listRunnableSubmissions(): Promise<AgentSubmission[]>;
  listUnreadySubmissions(): Promise<AgentSubmission[]>;
  listRunningSubmissions(): Promise<AgentSubmission[]>;
  listPendingSubmissionSettlements(): Promise<SubmissionSettlementObligation[]>;
  replaceSubmissionAttempt(
    attempt: SubmissionAttemptRef,
    nextAttemptId: string,
    lease?: { ownerId: string; leaseExpiresAt: number },
  ): Promise<AgentSubmission | null>;
  // 接纳
  admitDispatch(input: DispatchInput): Promise<AgentDispatchAdmission>;
  admitDirect(input: AgentSubmissionInput): Promise<AgentSubmission>;
  markSubmissionCanonicalReady(submissionId: string): Promise<AgentSubmission | null>;
  // 生命周期
  claimSubmission(claim: SubmissionClaimRef): Promise<AgentSubmission | null>;
  markSubmissionInputApplied(
    attempt: SubmissionAttemptRef,
    durability?: SubmissionDurability,
  ): Promise<boolean>;
  requestSessionAbort(sessionKey: string): Promise<string[]>;
  requeueSubmission(attempt: SubmissionAttemptRef): Promise<boolean>;
  reserveSubmissionSettlement(
    attempt: SubmissionAttemptRef,
    settlement: { recordId: string; record: SubmissionSettledRecord },
  ): Promise<SubmissionSettlementObligation | null>;
  finalizeSubmissionSettlement(
    attempt: SubmissionAttemptRef,
    recordId: string,
    options?: { errorMessage?: string },
  ): Promise<boolean>;
  completeSubmission(attempt: SubmissionAttemptRef): Promise<boolean>;
  failSubmission(attempt: SubmissionAttemptRef, error: unknown): Promise<boolean>;
  // 轮次边界加入
  claimJoinableSubmissions(
    host: SubmissionAttemptRef,
    agentName: string,
  ): Promise<AgentSubmission[]>;
  finalizeJoinedSubmission(host: SubmissionAttemptRef, submissionId: string): Promise<boolean>;
  revertJoiningSubmission(host: SubmissionAttemptRef, submissionId: string): Promise<boolean>;
  listJoinedSubmissions(hostSubmissionId: string): Promise<AgentSubmission[]>;
  // 租约管理
  renewLeases(ownerId: string, submissionIds: string[]): Promise<void>;
  listExpiredSubmissions(): Promise<AgentSubmission[]>;
}

辅助类型——AgentSubmission、SubmissionAttemptRef、SubmissionClaimRef、SubmissionDurability、SubmissionSettlementObligation、AgentDispatchAdmission、AgentDispatchReceipt、AgentSubmissionInput 和 DispatchInput——都从 @flue/runtime/adapter 导出。

查询方法：

• getSubmission — 提交，或 ID 未知时为 null。
• hasUnsettledSubmissions — 当任何提交排队、运行或 joining/joined 时为 true。
• listRunnableSubmissions — 每个会话中最旧的未结算提交的排队提交，按接纳顺序；每个会话最多一个可运行头部。
• listUnreadySubmissions — 没有规范就绪的排队提交，按接纳顺序。
• listRunningSubmissions — 所有运行中的提交，按接纳顺序。
• listPendingSubmissionSettlements — 已保留但尚未最终化的结算义务。
• replaceSubmissionAttempt — 恢复交接：原子地将运行中的提交移动到新的尝试 ID，增加 attemptCount，并在给定时安装新租约；当提交不在 attempt 下运行时不写入返回 null。

接纳方法：

• admitDispatch — 按提交 ID 键控的幂等接纳：精确重放返回已接纳的提交，相同 ID 不同载荷返回 { kind: 'conflict' }，匹配已保留回执的 ID 返回 { kind: 'retained_receipt' } 而不重新接纳。
• admitDirect — 将直接提示接纳为排队提交；对相同提交 ID 和载荷的精确重放是幂等的。
• markSubmissionCanonicalReady — 标记排队提交的规范对话已物化；排队时幂等，缺失或不再排队时返回 null。

生命周期方法：

• claimSubmission — 从排队到运行的原子比较并设置，仅当提交是其会话的可运行头部时；记录尝试、所有者、租约和开始时间。两个并发声明绝不能都成功。
• markSubmissionInputApplied — 在首次输入应用时一次性安装持久性预算（或默认值），盖上 inputAppliedAt 作为一次性守卫；在 attempt 拥有的运行中提交上门控。该标记是记账——规范流而非此时间戳才是输入是否持久化的真相。
• requestSessionAbort — 在会话中每个未结算的提交上盖上 abortRequestedAt（首次请求胜出）并返回它们的 ID。从不结算任何内容，从不更改 status；终端结算始终通过基于尝试的路径发生。
• requeueSubmission — 将运行中的提交返回排队进行干净的首次尝试，清除尝试、所有者、租约和持久性标记；仅在所有权上门控。
• reserveSubmissionSettlement — 原子地将精确的规范结算记录保留为义务（状态变为 terminalizing）；精确重放返回现有义务，冲突的记录身份或载荷返回 null。
• finalizeSubmissionSettlement — 在规范记录存在后最终化拥有的 terminalizing 提交；行的错误列镜像结算结果。
• completeSubmission / failSubmission — 结算拥有的运行中提交；过期的尝试或已结算的提交返回 false，第一个终端状态被保留。结算主机时原子地结算附加到它的每个 joined 提交并具有相同结果，并将任何未确认的 joining 残留者恢复为 queued。

轮次边界加入方法（繁忙时调度）：

• claimJoinableSubmissions — 原子地声明会话排队提交的连续前缀以吸收到主机的实时响应中（queued → joining，设置 joinedInto）；在主机仍在其尝试下运行时门控，因此被替换的尝试不声明任何内容。
• finalizeJoinedSubmission — 一旦投递的规范输入记录持久化就确认声明的加入（joining → joined）。
• revertJoiningSubmission — 将声明但未确认的加入交还队列（joining → queued）；仅在投递的规范输入记录不存在时合法。
• listJoinedSubmissions — 附加到主机的每个未结算加入，按接纳顺序。

租约方法：

• renewLeases — 延长列出的每个正在运行且由 ownerId 拥有的提交的租约到期时间（now + LEASE_DURATION_MS）；其他静默跳过。
• listExpiredSubmissions — 租约已到期的运行中提交；排队和已结算的提交从不返回。

导出常量：DURABILITY_DEFAULT_MAX_ATTEMPTS（10）、DURABILITY_DEFAULT_TIMEOUT_MS（3_600_000）、LEASE_DURATION_MS（30_000）。

ConversationStreamStore

按智能体实例的规范对话流：有序、追加式的 ConversationRecord 值批次，由单个受保护的生产者写入。流是唯一权威的对话记录——规范状态通过从头重放重建，适配器不得在会话行、快照或事件流中建模第二个对话记录。被拒绝的操作抛出 ConversationStreamStoreError 并保持流不变。

interface ConversationStreamStore {
  createStream(path: string, identity: ConversationStreamIdentity): Promise<void>;
  acquireProducer(path: string, producerId: string): Promise<ConversationProducerClaim>;
  append(input: {
    path: string;
    producerId: string;
    producerEpoch: number;
    incarnation: string;
    producerSequence: number;
    submission?: { submissionId: string; attemptId: string };
    records: readonly ConversationRecord[];
  }): Promise<{ offset: string }>;
  read(
    path: string,
    options?: { offset?: string; limit?: number },
  ): Promise<ConversationStreamReadResult>;
  getMeta(path: string): Promise<ConversationStreamMeta | null>;
  subscribe(path: string, listener: () => void): () => void;
  putFoldCheckpoint?(path: string, checkpoint: ConversationFoldCheckpoint): Promise<void>;
  getFoldCheckpoint?(
    path: string,
    options?: { atOrBefore?: string },
  ): Promise<ConversationFoldCheckpoint | null>;
}

• createStream — 如果不存在则创建流，铸造新的 incarnation ID。具有相同身份的竞态创建都成功；现有路径的冲突身份抛出。
• acquireProducer — 获取独占生产权：增加生产者 epoch，重置生产者序列，并返回携带 epoch、incarnation 和当前头部偏移量的声明。获取对所有先前生产者进行隔离。
• append — 在一个偏移量下追加一批记录。需要当前生产者 ID、epoch 和 incarnation，以及下一个预期的 producerSequence。已追加序列的精确重放返回原始偏移量；冲突的重放抛出。携带 submissionId/attemptId 的记录需要拥有它们的 submission 授权。批次中的每条记录一起持久化，全有或全无——部分写入会损坏对话图。
• read — options.offset 之后的严格批次（默认 '-1'，起始）。哨兵 'now' 不返回批次，当前头部作为 nextOffset。limit 被限制在 DEFAULT_READ_LIMIT（100）和 MAX_READ_LIMIT（1000）之间。超过头部的偏移量抛出；未知路径返回空的、最新的结果。
• getMeta — 流的身份、incarnation、头部偏移量和生产者状态，或未知路径的 null。
• subscribe — 注册路径的进程本地变更监听器；返回取消订阅函数。通知是尽力而为的进程内扇出，不是持久或跨进程信号。
• putFoldCheckpoint / getFoldCheckpoint — 可选的折叠检查点能力：每个路径一个持久的序列化折叠快照，每次写入时被取代，因此加载只折叠自其以来追加的后缀，而不是从源头重放流。检查点是日志之上的缓存，从不权威——运行时在加载时验证其格式版本、incarnation 和偏移量，任何不匹配时静默通过重放重建。没有这对方法的适配器保持完全功能；运行时降级为完整重放并每个路径警告一次。实现必须是撕裂写安全的：部分持久化的检查点必须读回为不存在或读取失败，永远不能是看似合理的 blob。带 atOrBefore 的 getFoldCheckpoint 仅在其偏移量在界限之内或之前时返回检查点。

偏移量是按流排序的不透明字符串；formatOffset 和 parseOffset 在偏移量字符串和整数序列号之间转换。defineSqlConversationStreamStore(dialect: SqlConversationDialect) 在异步 SQL 后端上构建完整的 ConversationStreamStore——Postgres、libSQL 和 MySQL 适配器共享一个隔离实现，仅在方言常量上不同。InMemoryConversationStreamStore 和 StreamListenerRegistry 作为参考构建块导出。

AttachmentStore

通过 AttachmentRef 引用的规范对话记录的不可变附件字节——存储身份和完整性元数据（id、mimeType、size、digest，可选 filename），不是下载 URL。filename 是展示元数据，不参与身份比较。

interface AttachmentStore {
  put(input: PutAttachmentInput): Promise<void>;
  get(input: GetAttachmentInput): Promise<StoredAttachment | null>;
}

• put — 在流中存储附件 ID 的字节。幂等：精确的重新 put（相同 ref、字节和对话）成功，包括并发的精确 put。用不同内容、元数据或所有权重用 ID 会抛出 AttachmentConflictError。字节针对 ref 的 size 和 digest 验证；不匹配抛出 AttachmentIntegrityError。
• get — 存储的附件和字节，或 ID 未知或 conversationId 不匹配时为 null。读取时验证完整性。

辅助函数：createAttachmentRef 构建 ref（计算 SHA-256 digest），verifyAttachmentBytes 检查字节是否匹配 ref，sameAttachmentRef 比较 ref 忽略 filename，attachmentBytesEqual 和 copyAttachmentBytes 操作字节数组，InMemoryAttachmentStore 是完整的参考实现。

跨切面要求

在所有三个存储中成立的规则；契约套件测试每一个。

• 幂等接纳。 接纳、追加、put 或结算保留的精确重放返回原始结果；相同身份不同内容是冲突，永远不是静默覆盖。
• 受保护的生产者声明。 每个对话流最多一个活跃生产者。acquireProducer 使所有先前声明无效；携带过期 epoch 或 incarnation 的追加被拒绝。提交拥有的追加还需要写入尝试持久地拥有提交。
• 追加式流。 规范记录从不更新或重写。批次在单个偏移量下是全有或全无的，偏移量严格有序。
• 第一个终端状态胜出。 已结算提交的结果从不被覆盖——过期的尝试从结算方法观察到 false。
• 可观察的原子性。 当方法被描述为原子时，并发调用者绝不能都观察到成功；是通过事务、条件更新还是唯一索引实现取决于适配器。
• 格式版本标记。 适配器首次创建存储时持久记录其格式版本（当前版本：FLUE_FORMAT_VERSION，1），当打开记录了未知或更新版本的存储时抛出 PersistedFormatVersionError——在读取或写入任何数据之前。assertSupportedFlueFormatVersion(storedVersion) 执行检查。格式是仅重置的：记录了其他版本的存储被清空，从不原地迁移。内置 SQL 适配器用单行 flue_meta 键/值表（键 'format_version'）实现标记；非 SQL 适配器原生实现相同的义务。

契约测试套件

@flue/runtime/test-utils 中的三个 vitest 套件是验收标准：当所有三个套件对其存储通过时，适配器就是正确的。每个函数注册一个 describe 块；每个测试从 backend.create() 接收新鲜的存储，backend.cleanup?() 在每个测试后运行。

import {
  defineAttachmentStoreContractTests,
  defineConversationStreamStoreContractTests,
  defineStoreContractTests,
} from '@flue/runtime/test-utils';

defineStoreContractTests('My backend', {
  async create() {
    return mySubmissionStore;
  },
  async cleanup() {
    /* 关闭连接，删除临时状态 */
  },
});

• defineStoreContractTests(label, backend) — AgentSubmissionStore 套件：接纳、规范就绪、队列排序、声明、生命周期转换、中止、结算义务、持久性标记、尝试替换、租约和轮次边界加入。backend.create() 返回 AgentSubmissionStore。可选的 backend.formatVersion 组给套件对持久化格式版本标记的原始访问（open() 返回新鲜的、未迁移的存储句柄，带 migrate、readStamp、writeStamp 和 deleteStamp）并启用格式版本标记测试。
• defineConversationStreamStoreContractTests(label, backend) — ConversationStreamStore 套件：竞态创建、有序原子批次、幂等和冲突重放、生产者隔离、提交拥有的追加授权和读取。backend.create() 返回 { stream, submissionStore? }；提交存储是授权测试所必需的。也可从 @flue/runtime/test-utils/conversation-stream 导入。
• defineAttachmentStoreContractTests(label, backend) — AttachmentStore 套件：字节往返、幂等和并发精确 put、身份重用时的冲突错误和完整性错误。backend.create() 返回 AttachmentStore。也可从 @flue/runtime/test-utils/attachment-store 导入。

适配器辅助函数

从 @flue/runtime/adapter 导出的纯辅助函数，内置适配器使用，自定义适配器也可使用：

• admitSubmissionWithBackend(input, backend) — 面向行后端的共享接纳算法（回执检查、附件准备、插入或忽略、回读、幂等重放与冲突比较、块采用）。调用者拥有事务范围；当每个 SubmissionAdmissionBackend 回调是同步的时，结果同步返回，适配同步事务包装器。
• isSubmissionPayload(input, ctx) — 针对存储的提交元数据（SubmissionPayloadContext）验证解析的 JSON 载荷。
• parseAcceptedAt(value, label) — 将 ISO 时间戳解析为 epoch 毫秒；无效值时抛出。
• clampLimit(limit, defaultLimit, maxLimit) — 无效或非正限制时回退到默认值，上限为最大值。
• createSessionStorageKey(agentName, instanceId, harness, session) / parseSessionStorageKey(key) — 序列化和解析会话通道身份，该身份隔离队列排序、中止和尝试所有权。外部提交始终使用 SUBMISSION_HARNESS_NAME 和 SUBMISSION_SESSION_NAME（均为 'default'）。
• createDispatchAgentSubmissionInput(input) — 将 DispatchInput 转换为持久化的 AgentSubmissionInput 形状。
• prepareSubmissionAttachments、hydratePersistedSubmissionAttachments、matchesPersistedSubmissionAttachments、sameSubmissionChunks — 超大行安全载荷存储的附件分块，按提交 ID 键控（SubmissionChunkRow、SubmissionChunkStore）。

适配器表面刻意狭窄：存储接口、词汇类型和纯辅助函数——没有运行时编排、Provider 管道或生成入口内部。错误类（ConversationStreamStoreError、AttachmentConflictError、AttachmentIntegrityError、PersistedFormatVersionError）在错误中有文档。