07 · Agent 的生命周期是什么？ - Agent Lifecycle

所属知识库：Agent 全栈技术地图
所属领域：02 · Agent
所属章节：01 · Agent 是什么？它如何工作？ - Fundamentals
文档类型：技术原理与工程实践研究
版本：v1.1
更新日期：2026 年 10 月 28 日
核心问题：Agent 定义、发布版本、运行实例、会话和任务执行分别具有什么生命周期？运行状态如何通过可信事件和并发控制改变？检查点如何持久化并恢复任务？重试、人工审批、外部副作用、版本迁移与终止条件如何形成可验证的运行契约？

专题导读

Agent 的生命周期并非单一的“创建-运行-结束”。一份 Agent 定义可能经历数十次版本发布；一个版本可以服务大量任务；一个任务可能跨多个进程、等待数天并经历多次恢复；任务完成后，产物、审计记录与记忆仍有各自的保留周期。生命周期研究首先需要确定被管理的对象，再研究对象之间的状态转移、持久化事实和执行约束。

本文保留 v1.0 的三层生命周期、对象边界、终止条件和工程风险分析，重点将状态机、检查点、工具副作用和恢复机制推进到可实现、可测试的技术层次。所提出的状态名称与 SQLite 原型属于本文的参考设计，不代表所有 Agent 框架的统一规范。文中对 LangGraph 与 Temporal 的比较基于公开的稳定架构概念，具体版本行为仍需在实际选型时对照官方文档和源码核验。

一、生命周期的对象、边界与时间尺度

1.1 三条主生命周期

定义与交付生命周期（Definition / Delivery）：需求与适用性判断、角色和能力设计、配置与代码构建、评测、发布、维护、回滚、退役。其对象是可复用的 Agent Definition 及其不可变 Release，通常以天、周、月计。

实例生命周期（Instance）：运行环境创建、资源绑定、就绪、执行或空闲、暂停、资源释放。其对象是实际承载任务的进程、容器或执行实体；一个实例可承载多个 Run，一个 Run 也可能跨多个实例恢复。实例终止不必然意味着 Run 失败。

任务生命周期（Run / Task）：接纳任务、初始化、执行、等待、暂停、恢复、完成、失败、取消或超时。其对象是具有目标、预算、状态和可追踪 ID 的一次业务执行；可能跨越多个 Worker 与进程。

会话、检查点、记忆、知识索引、外部副作用、任务产物分别有独立的数据生命周期。不能从 Run 的终态直接推断这些对象应被删除，也不能把删除 Agent Definition 视为撤销已经发生的外部操作。

1.2 核心对象模型

对象	标识	核心字段或引用	生命周期关注点
Agent Definition	agent_id	角色、指令、工具、策略	定义更新与退役
Agent Release	release_id	定义哈希、模型策略、工具 Schema、策略版本	不可变性与兼容性
Agent Instance	instance_id	进程/容器、租约、执行环境	资源分配与回收
Session	session_id	用户、租户、对话范围	多轮连续性与隔离
Run	run_id	release、任务、状态、预算、截止时间	执行、等待、恢复、终止
Step / Attempt	step_id / attempt_id	行动、尝试次数、工具结果	局部重试与执行事实
Checkpoint	checkpoint_id	状态版本、步骤游标、操作引用	崩溃恢复
Operation	operation_id	工具调用意图、幂等键、外部结果	副作用去重与核对
Artifact	artifact_id	内容地址、版本、访问策略	交付与保留

关系上，一个 Definition 可拥有多个 Release；一个 Release 可绑定多个 Run；一个 Session 可关联多个 Run；一个 Run 可跨多个 Attempt、Checkpoint 和 Worker。run_id 是业务执行身份，instance_id 是执行资源身份。当 Worker 崩溃后由另一个 Worker 接管，run_id 不应被迫改变。

1.3 识别对象混淆的反例

假设用户在同一 Session 内提交两次研究任务。两次任务共用 Release，但各自有独立 Run、预算、步骤和结果。第二次任务可以在授权条件下使用会话上下文，却不应自动继承第一次任务的临时发布许可。如果只用 session_id 保存任务状态，就可能发生任务间预算混算、审批误复用和重复写入。若只用 instance_id 标识任务，进程重启后又可能丢失业务连续性。

二、开发、发布与演进生命周期

2.1 从任务契约到可发布定义

设计前先判断目标是否真的需要 Agent：固定步骤和明确规则通常可以由传统程序或工作流承担；需要在不确定环境中自主选择下一步动作时，Agent 才可能带来额外价值。任务契约至少应明确输入、输出、验收条件、可执行行动、可访问数据、预算、截止时间、失败和人工介入条件。随后将角色、指令、模型路由、工具声明、状态结构、记忆策略、知识来源和权限策略转化为可审查的定义。

发布流程可写成 DRAFT → BUILT → EVALUATED → RELEASED → DEPRECATED → RETIRED。这是定义版本的状态，不应与 Run 的状态机混用。Release 应尽可能包含不可变的配置快照与内容哈希；密钥和运行时凭证不应固化到 Release 中。开发、测试和生产环境可采用不同凭证与服务地址，但环境绑定必须可追溯。

2.2 发布门禁与变更归因

发布前至少检查：配置 Schema、工具契约、权限拒绝路径、正常任务、无答案任务、工具失败、超时、预算耗尽、敏感数据处理和输出质量。对于模型行为的随机性，不能以单次演示替代评测；应保留输入集、样本数量、结果分布和失败案例。发布清单记录定义哈希、指令版本、模型策略、工具 Schema、技能版本、知识索引引用、策略版本、运行时兼容范围与评测证据。

模型别名、第三方 API 和动态知识源可能发生外部变化，因此“固定 Release”不等于完全可复现。可复现性需要记录可取得的实际模型标识、调用参数、外部证据版本、时间与环境信息，并承认非确定性及服务方不可控变化。

2.3 在途任务与退役

升级可以采用：旧 Run 固定旧 Release、新 Run 使用新 Release；在检查点执行显式迁移；终止旧 Run 后新建 Run；紧急安全修复时进行受控强制迁移。迁移必须校验状态 Schema、工具契约、待处理操作和权限。回滚 Release 只影响后续执行，不会撤销已经发出的邮件或已经写入的外部数据。退役还应明确新任务是否接纳、旧任务如何排空、检查点保留多久、产物与审计如何处理。

三、任务状态机：从状态图到原子转移协议

3.1 参考状态模型

CREATED → VALIDATING → QUEUED → INITIALIZING → RUNNING
                         │                     ├── WAITING ───┐
                         │                     ├── PAUSED ────┤
                         │                     ├── CANCELLING → CANCELLED
                         │                     ├── COMPLETED  │
                         │                     ├── FAILED     │
                         │                     └── TIMED_OUT  │
                         └── FAILED                          │
                                           RUNNING ◄─────────┘

上图是简化示意。WAITING 可以表示等待审批、异步工具或用户输入；PAUSED 通常是主动暂停；CANCELLING 表示已接收取消请求但尚未确认清理完成。终态包括 COMPLETED、FAILED、CANCELLED、TIMED_OUT。实际合法转移应通过明确的转移表表达，而不是依赖示意图推断。

3.2 状态、原因、事件是三个不同概念

status=WAITING 是机器可判定的当前状态，waiting_reason=human_approval 是原因，APPROVAL_REQUESTED 是导致状态变化的事件。将每一种等待原因都创建成新状态，会使状态机组合爆炸；反过来只存一个 status 而没有事件，又难以回答“谁、何时、为什么把任务改成等待”。

建议将 Run 作为状态聚合根，所有状态变更由权威运行时提交。模型生成的“我已完成”只能作为候选内容，不能直接更新 COMPLETED；完成条件应由输出校验、产物存在性、业务验收或其他可执行规则确认。

3.3 并发状态转移的具体协议

假设两个 Worker 同时读取 run_042 的 status=RUNNING, version=7：Worker A 想将它改为 WAITING，Worker B 收到取消事件，想改为 CANCELLING。如果两者直接覆盖数据库行，后提交者会抹去前者的事实。

一种最小可行的解决方案是乐观并发控制（Optimistic Concurrency Control）：

1. 读取 run_id、当前状态及 version。
2. 校验 old_status → new_status 是否在合法转移表内。
3. 在事务中执行条件更新：UPDATE runs SET status=?, version=version+1 WHERE run_id=? AND version=? AND status=?。
4. 仅当受影响行数为 1 时，提交同事务内的状态事件；否则回滚并重新读取。
5. 对重复到达的外部事件，通过唯一 event_id 或请求键去重，避免重复生效。

BEGIN IMMEDIATE;
UPDATE runs
   SET status = :new_status,
       version = version + 1
 WHERE run_id = :run_id
   AND version = :expected_version
   AND status = :expected_status;
-- 只有更新成功才写入同事务的 run_events
COMMIT;

BEGIN IMMEDIATE 是 SQLite 的写事务语义；在 PostgreSQL 等数据库中可使用条件更新、行锁或其他事务机制。状态变更与事件记录必须保持一致：如果状态已经更新但事件丢失，审计链会断裂；如果事件已写入但状态未更新，恢复时又会出现矛盾。原子事务解决本地状态和本地事件的一致性，但不能使数据库事务与远端邮件服务成为一个原子事务。

3.4 终态、取消竞争与租约

终态原则上不可随意“复活”；重试、恢复或分叉应通过明确的新 Attempt、子 Run 或恢复操作表示。取消与工具完成可能同时发生：取消请求到达时，远端操作也许已经提交，因此应分别记录 cancel_requested_at 和已确认的工具副作用。取消成功只表示运行时停止继续调度，不代表外部副作用已回滚。

多 Worker 环境还需要执行租约（lease）或所有权标识：某个 Worker 领取 Run 后持有带期限的执行权；租约过期可以重新分配。租约本身无法防止旧 Worker 在网络恢复后继续写入，因此需要 fencing token 或基于递增版本的写入检查。本文后面的 SQLite 原型只实现单进程原子状态与检查点，不包含分布式租约。

四、Agent Loop 如何驱动生命周期

一次运行循环可以形式化为：读取权威状态 → 装配当前上下文 → 模型产生候选行动 → 校验结构与权限 → 执行或等待 → 保存观察结果 → 更新任务状态 → 判断终止。模型负责提出行动，Runtime 负责执行与状态事实，Tool Adapter 负责将外部结果转成可处理的观察，策略执行层负责授权检查。四者不应混为一体。

循环阶段	输入	输出	可能触发的生命周期变化
任务接纳	目标、Release、发起者、预算	已校验 Run	CREATED → QUEUED 或 FAILED
上下文装配	Run 状态、历史、检索结果	模型输入	不一定变更 Run 状态
模型决策	当前输入、模型策略	候选答案/工具调用	RUNNING 保持或转 WAITING
行动授权	操作、参数、身份、策略	允许/拒绝/待审批	RUNNING → WAITING / FAILED
工具执行	operation_id、工具参数	结果或不确定状态	RUNNING / WAITING / FAILED
状态提交	观察、步骤结果、预算	新 Checkpoint、事件	继续 RUNNING 或终态

等待不是忙等。 当需要人工审批或长时间异步结果时，运行时应持久化待处理动作、等待原因和关联 ID，提交 WAITING，然后释放执行资源。外部事件到达后，先核验事件身份、对象与有效期，再恢复 Run。把等待实现为一个始终占用线程的 sleep，会使长任务难以扩展，也无法可靠跨进程重启。

终止条件分为业务条件（输出达到验收标准、确无可用证据、用户撤销）与硬性限制（步骤数、时间、Token、金额、工具次数、权限或安全策略）。预算应在真正发起新行动前检查，已耗费预算必须持久化，防止崩溃恢复后重新获得一份免费额度。部分完成的任务可输出中间产物，但不能将部分结果冒充完整交付。

五、检查点、持久化与执行恢复

5.1 运行时状态由什么构成

可恢复状态至少包括：Run 与 Release 身份、状态版本、当前步骤游标、已确认的步骤结果、待执行动作、操作账本引用、预算使用、等待事件关联、产物引用和外部资源位置。模型上下文通常是从这些权威事实和必要历史重新装配的视图，不应把“完整提示词字符串”当作唯一恢复载体。

{
  "run_id": "run_042",
  "release_id": "research-assistant@1.3.2",
  "state_version": 17,
  "cursor": "publish_report",
  "status": "WAITING",
  "waiting": {"kind": "human_approval", "approval_id": "apr_91"},
  "completed_steps": ["scope", "search", "draft"],
  "budget_used": {"steps": 17, "tool_calls": 9},
  "pending_operation_id": "op_19",
  "artifact_refs": ["artifact://reports/draft-042"]
}

这是说明性结构，不是某个框架的原生 API。敏感内容应存储在受控位置，检查点尽量保存引用而非明文凭证；恢复时重新获取有效授权。

5.2 快照、事件重放与混合模式

策略	写入机制	恢复机制	主要代价
全量快照	在安全边界写完整状态	读取最近有效快照	写入体积与频率
事件日志	追加领域事件	从初始状态重放事件	重放成本、事件版本演进
混合策略	周期性快照 + 后续事件	读取快照并重放增量	两类存储的一致性

选择检查点频率需要在恢复损失窗口、存储成本和执行吞吐量之间取舍。每步写入可减少重做范围，但会增加写放大；只在任务结束写入几乎不能支持中途恢复。对于外部不可逆操作，应在执行前持久化操作意图，并在结果确认后持久化执行事实；即使使用每步快照，也不能把外部服务的真实状态当作本地事务的一部分。

5.3 恢复算法及其不变量

一个安全恢复流程应按以下顺序执行：

1. 定位执行身份：读取 Run、绑定 Release 和最近有效状态版本，确认任务非终态且具备恢复资格。
2. 重建状态：读取 Checkpoint，必要时重放后续事件，验证数据结构与版本兼容性。
3. 核对外部事实：检查未确认的工具操作，区分明确失败、明确成功和结果未知。
4. 重新授权：检查当前用户/租户、凭证、权限策略与审批有效期，不能复用已经撤销的权限。
5. 确定恢复点：从已确认步骤之后继续；对不确定的副作用进入核对流程，禁止盲目重复写操作。
6. 原子接管执行权：更新版本或租约，再继续调度；记录恢复来源、恢复时间和执行主体。

关键不变量包括：已确认的业务副作用不被重复提交；同一 Run 的预算不因重启归零；状态版本单调递增；旧审批不能授权不同参数的动作；恢复不会无声替换 Release；同一外部事件重复到达不应导致两次状态转移。

5.4 崩溃窗口分析

崩溃时点	已持久化事实	恢复动作
模型调用前	上一 Checkpoint	重新装配上下文并调用模型；注意模型结果可能变化
工具意图持久化后、发送前	PREPARED 操作	确认尚未发送后调度
请求发送后、响应前	DISPATCHED / UNKNOWN	查询远端结果或人工核对；不直接重复非幂等动作
远端成功后、本地提交前	可能只有操作 ID	通过幂等键或外部查询恢复结果
Checkpoint 提交后、响应用户前	已确认状态和产物	返回已存在结果，不重新执行

“从检查点恢复”不意味着可以在任意 Python 指令处继续执行；通常需要以步骤、节点、工作流事件或其他明确边界重新进入。不同框架的恢复粒度不同，且某些节点可能重新执行，因此节点内部副作用仍需幂等设计。

六、工具执行协议、重试与副作用一致性

6.1 业务操作与执行尝试必须分离

一个 operation_id 代表业务上希望发生的一次动作；一个 attempt_id 代表某次网络或工具执行尝试。同一 operation 可以有多次 attempt，但不能因此产生多次业务效果。操作记录建议包含：操作类型、参数哈希、调用主体、审批 ID、幂等键、状态、尝试次数、外部资源 ID、结果摘要和核对时间。

PREPARED → DISPATCHED ──► SUCCEEDED
                  │      └► FAILED
                  └──────► UNKNOWN → RECONCILING
                                          ├► SUCCEEDED
                                          ├► FAILED
                                          └► MANUAL_REVIEW

UNKNOWN 不是失败的同义词：它表示运行时缺乏足够证据判断远端动作是否生效。尤其是邮件发送、付款、发布和删除等操作，不能在 UNKNOWN 时直接重新执行。

6.2 可操作的执行协议

1. 根据 Run、步骤和业务参数生成稳定 operation_id，将操作意图与参数摘要持久化。
2. 检查当前权限与审批是否绑定同一操作和参数。
3. 调用支持幂等键的远端服务时，传递稳定幂等键；若服务不支持，应设计查询、去重或人工核对路径。
4. 请求发出后记录 attempt 与可观察结果；超时不自动等于失败。
5. 若结果未知，优先查询远端状态；无法核实时停止危险重试，转人工处理。
6. 确认成功后持久化外部资源 ID 和结果，再推进后续步骤。

对于只读搜索，可采用带上限和退避的重试；对按稳定 ID 的幂等更新，可使用远端幂等机制；对非幂等操作，应优先保证去重与结果核对。所谓 exactly-once 往往只能在明确的业务边界和外部服务支持下实现效果去重，不能把跨服务网络调用整体描述为天然恰好执行一次。

6.3 Outbox、补偿与不确定状态

当本地数据库提交与外部请求无法处于同一事务时，可以使用事务性 Outbox：先在本地事务中保存操作意图和待发送记录，再由 Dispatcher 发送。它减少“状态已提交但发送意图丢失”的风险，却不消除“远端已执行、响应丢失”的不确定性；仍需远端幂等或核对机制。补偿操作（例如撤销发布）属于新的业务动作，不是数据库意义上的回滚，可能失败且不一定能恢复到原始世界状态。

七、代表性技术路线：LangGraph 与 Temporal

7.1 为什么比较两者

LangGraph 主要提供基于图与状态的 Agent 编排能力；Temporal 主要提供持久化工作流执行基础设施。二者可用于同一系统的不同层次，不应被视为同类产品的简单替代。比较重点是它们如何组织执行边界、持久化状态、处理中断和恢复、管理重试与副作用。

7.2 LangGraph：图状态与检查点

LangGraph 的核心思路是将执行表示为图节点与状态更新。节点读取图状态，执行逻辑并返回更新；状态通过图运行过程逐步演进。持久化能力通常由 Checkpointer 与 Thread 标识关联；同一 Thread 可关联一系列检查点。人工介入可以通过中断机制暂停执行，并在后续提供恢复输入。

理解其生命周期时应关注四个问题：图状态是什么、检查点在什么执行边界生成、恢复从哪个边界重新进入、节点内部的副作用是否会再次发生。 对需要审批的步骤，必须把“请求审批”和“执行已批准动作”分离，防止恢复时意外重复执行。图状态持久化并不自动解决远端 API 的幂等性，也不能代替授权策略。

示意执行链：thread_id → checkpoint → graph state → interrupt → external approval → resume → next node。这是机制级抽象，非对某个 SDK 版本接口签名的保证。实际开发应核验对应 LangGraph 版本的持久化粒度、中断语义、任务重放和副作用处理建议。

7.3 Temporal：事件历史与确定性重放

Temporal 以 Workflow Execution 为长期业务执行身份，服务端持久化工作流事件历史；Worker 执行工作流代码，并在需要时依据历史重放重建工作流逻辑状态。Workflow 的逻辑要求遵循确定性约束，外部网络、数据库和其他不确定副作用通常放在 Activity 等受控执行边界中。Activity 可按配置重试，执行超时或 Worker 故障不意味着外部副作用必然未发生。

其重要机制不是“保存 Python 对象然后从任意一行恢复”，而是根据历史事件重建工作流决策进程。因此工作流代码变更需要考虑历史兼容性；Activity 的重试需要幂等设计；长期等待可通过工作流消息、信号或其他受支持的交互机制继续推进。具体名称、API 与版本策略应按实际 SDK 文档核对。

7.4 技术比较与选型

维度	LangGraph（概念层）	Temporal（概念层）	生命周期启示
执行模型	图节点与状态更新	持久化工作流与事件历史	先明确业务执行边界
持久化重心	图状态、检查点、Thread	Workflow History 与重放	恢复粒度不同
人工等待	中断与恢复	长期工作流等待与外部消息	等待不应依赖常驻线程
故障恢复	依据检查点恢复图执行	依据历史重放工作流逻辑	必须理解重新执行边界
副作用	节点/工具需单独处理幂等	Activity 需单独处理幂等	框架不会自动撤销外部动作
主要适用重点	Agent 状态图、模型与工具编排	长期可靠业务流程与执行基础设施	可组合使用，不必二选一

证据边界：本节是对公开架构思想的技术综合，不声称已在 2026 年 10 月 28 日逐项核验最新 SDK、源码或部署版本；更不声称两框架的所有实现行为都与上述抽象完全一致。

八、版本、权限与数据的交叉生命周期

8.1 版本迁移

在途 Run 应默认绑定创建时的 Release。迁移需显式执行兼容性检查：状态 Schema 是否可读、步骤游标是否有效、待执行工具是否仍存在、审批绑定是否仍有效、预算和外部操作记录是否保持。若发现旧版安全缺陷，允许紧急暂停或迁移，但应记录理由、操作者和补救路径。版本迁移成功不意味着旧任务的外部副作用自动回滚。

8.2 权限随时间变化

Agent Definition 声明工具能力，不等于每个用户或 Run 都拥有执行权限。授权至少受当前发起者身份、租户、资源、动作参数和策略版本约束。敏感操作前必须重新授权，特别是在审批等待、恢复、会话续接和跨环境迁移之后。审批应绑定具体操作及参数摘要、审批人、有效期和允许范围；外部网页、检索文档和工具返回文本不能改变可信策略。

8.3 数据保留与删除

Session 历史、Checkpoint、Memory、知识索引、执行轨迹、产物和操作审计的保留周期应分别设计。删除 Run 的可恢复状态可能影响故障排查；长期保留原始工具响应又可能扩大敏感信息风险。建议通过数据分类、访问控制、加密、脱敏、保留期限和删除传播处理这些问题。具体法律义务取决于业务所在地和数据类型，本篇不提供通用法律结论。

九、贯穿案例：研究助理 Agent 的一次可恢复运行

研究助理 Release research-assistant@1.3.2 被配置为：可读搜索、文档解析、草稿生成和受控发布；最多 30 个行动步骤；发布前必须由发起者审批；每次确认的步骤结束后写入检查点。

用户要求比较三种技术路线并形成报告。系统创建 run_042，校验任务后进入 RUNNING；步骤 scope 确定研究范围，search 取得证据，draft 写入草稿产物。此时 Runtime 生成发布操作意图 op_19，把草稿哈希和审批 ID 绑定，原子保存 WAITING 状态及检查点，Worker 退出。

第二天另一个 Worker 收到审批事件。系统读取 Run 和 Checkpoint，核对审批人与草稿哈希，重新检查当前发布权限。若报告在等待期间被修改，旧审批不应自动授权新版本。核验通过后恢复执行，使用稳定的操作 ID 调用发布服务。若请求超时，先查询发布状态；确认成功后记录外部资源 ID，并依据产物可访问性和验收规则将 Run 标记为 COMPLETED。

开发者之后修正引用规则并发布 1.3.3，新 Run 默认使用新版本；历史 run_042 继续保留 1.3.2 的引用。这个案例同时展示：等待可跨进程、恢复需重新授权、审批绑定具体参数、外部成功不能仅凭模型自报、Release 与 Run 的寿命不同。

十、工程实现：SQLite 最小可恢复执行器

10.1 实现范围与边界

下面的 Python 3 标准库示例实现：持久化 Run、合法状态转移、状态版本 CAS、事件日志、检查点、等待审批、重启后恢复、审批事件去重，以及模拟外部发布操作的基本幂等。示例不调用真实模型，不实现分布式 Worker、真实权限服务或远端事务；其目的是验证生命周期控制机制，而不是演示完整 Agent 智能。

将代码保存为 lifecycle_demo.py，运行 python3 lifecycle_demo.py。数据库使用临时文件，示例在创建第一个执行器并进入 WAITING 后，关闭连接、重新创建执行器，模拟进程退出后的恢复。下面的示例使用 SQLite 单文件事务，所有操作与状态修改均在同一个本地数据库中，因此不能据此推断真实外部服务调用已经实现端到端 exactly-once。

from __future__ import annotations
import json
import os
import sqlite3
import tempfile
from dataclasses import dataclass

ALLOWED = {
    "CREATED": {"VALIDATING", "CANCELLED"},
    "VALIDATING": {"QUEUED", "FAILED"},
    "QUEUED": {"INITIALIZING", "CANCELLED"},
    "INITIALIZING": {"RUNNING", "FAILED"},
    "RUNNING": {"WAITING", "PAUSED", "COMPLETED", "FAILED", "CANCELLED"},
    "WAITING": {"RUNNING", "CANCELLED"},
    "PAUSED": {"RUNNING", "CANCELLED"},
    "COMPLETED": set(), "FAILED": set(), "CANCELLED": set(),
}

@dataclass(frozen=True)
class Snapshot:
    run_id: str
    status: str
    version: int
    cursor: str

class Runtime:
    def __init__(self, path: str):
        self.db = sqlite3.connect(path, isolation_level=None)
        self.db.execute("PRAGMA foreign_keys=ON")
        self.db.executescript("""
        CREATE TABLE IF NOT EXISTS runs (
          id TEXT PRIMARY KEY, release TEXT NOT NULL,
          status TEXT NOT NULL, version INTEGER NOT NULL DEFAULT 0,
          cursor TEXT NOT NULL DEFAULT 'scope');
        CREATE TABLE IF NOT EXISTS events (
          event_id TEXT PRIMARY KEY, run_id TEXT NOT NULL,
          kind TEXT NOT NULL, detail TEXT NOT NULL,
          FOREIGN KEY(run_id) REFERENCES runs(id));
        CREATE TABLE IF NOT EXISTS checkpoints (
          run_id TEXT NOT NULL, version INTEGER NOT NULL,
          payload TEXT NOT NULL, PRIMARY KEY(run_id, version));
        CREATE TABLE IF NOT EXISTS operations (
          operation_id TEXT PRIMARY KEY, run_id TEXT NOT NULL,
          status TEXT NOT NULL, external_id TEXT);
        """)

    def close(self):
        self.db.close()

    def create(self, run_id: str, release: str):
        with self.db:
            self.db.execute("INSERT INTO runs(id, release, status) VALUES(?,?,'CREATED')",
                            (run_id, release))
            self.db.execute("INSERT INTO events VALUES(?,?,?,?)",
                            (f"{run_id}:created", run_id, "RUN_CREATED", "{}"))
            self.db.execute("INSERT INTO checkpoints VALUES(?,?,?)",
                            (run_id, 0, json.dumps({"cursor": "scope"})))

    def get(self, run_id: str) -> Snapshot:
        row = self.db.execute("SELECT id,status,version,cursor FROM runs WHERE id=?",
                              (run_id,)).fetchone()
        if row is None:
            raise KeyError(run_id)
        return Snapshot(*row)

    def move(self, run_id: str, target: str, event_id: str,
             cursor: str | None = None) -> Snapshot:
        self.db.execute("BEGIN IMMEDIATE")
        try:
            old = self.get(run_id)
            if self.db.execute("SELECT 1 FROM events WHERE event_id=?",
                               (event_id,)).fetchone():
                self.db.execute("COMMIT")
                return old  # duplicate event is a no-op
            if target not in ALLOWED[old.status]:
                raise ValueError(f"illegal transition {old.status} -> {target}")
            next_cursor = old.cursor if cursor is None else cursor
            changed = self.db.execute(
                "UPDATE runs SET status=?, version=version+1, cursor=? "
                "WHERE id=? AND version=? AND status=?",
                (target, next_cursor, run_id, old.version, old.status)).rowcount
            if changed != 1:
                raise RuntimeError("concurrent state change")
            new = self.get(run_id)
            self.db.execute("INSERT INTO events VALUES(?,?,?,?)",
                            (event_id, run_id, f"RUN_{target}",
                             json.dumps({"from": old.status, "to": target})))
            self.db.execute("INSERT INTO checkpoints VALUES(?,?,?)",
                            (run_id, new.version,
                             json.dumps({"cursor": new.cursor, "status": new.status})))
            self.db.execute("COMMIT")
            return new
        except Exception:
            self.db.execute("ROLLBACK")
            raise

    def publish_once(self, run_id: str, operation_id: str) -> str:
        # Local stand-in for an external service that supports idempotency keys.
        self.db.execute("BEGIN IMMEDIATE")
        try:
            row = self.db.execute(
                "SELECT status,external_id FROM operations WHERE operation_id=?",
                (operation_id,)).fetchone()
            if row:
                if row[0] != "SUCCEEDED":
                    raise RuntimeError("operation requires reconciliation")
                self.db.execute("COMMIT")
                return row[1]
            external_id = f"published:{operation_id}"
            self.db.execute("INSERT INTO operations VALUES(?,?,?,?)",
                            (operation_id, run_id, "SUCCEEDED", external_id))
            self.db.execute("COMMIT")
            return external_id
        except Exception:
            self.db.execute("ROLLBACK")
            raise

def test_demo(path: str):
    r = Runtime(path)
    r.create("run_042", "research-assistant@1.3.2")
    for status in ("VALIDATING", "QUEUED", "INITIALIZING", "RUNNING"):
        r.move("run_042", status, f"ev:{status}")
    waiting = r.move("run_042", "WAITING", "ev:approval-requested",
                     cursor="publish_report")
    assert waiting.status == "WAITING" and waiting.cursor == "publish_report"
    r.close()  # simulate stopping the process

    r2 = Runtime(path)  # new process can reconstruct from persisted state
    assert r2.get("run_042").status == "WAITING"
    assert r2.move("run_042", "WAITING", "ev:approval-requested") == r2.get("run_042")
    r2.move("run_042", "RUNNING", "ev:approval-granted")
    first = r2.publish_once("run_042", "op:publish:042")
    second = r2.publish_once("run_042", "op:publish:042")
    assert first == second
    r2.move("run_042", "COMPLETED", "ev:completed", cursor="done")
    assert r2.get("run_042").status == "COMPLETED"
    try:
        r2.move("run_042", "RUNNING", "ev:illegal-restart")
    except ValueError:
        pass
    else:
        raise AssertionError("terminal state must be immutable")
    assert r2.db.execute("SELECT count(*) FROM operations").fetchone()[0] == 1
    assert r2.db.execute("SELECT count(*) FROM checkpoints WHERE run_id='run_042'").fetchone()[0] >= 7
    r2.close()

if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as directory:
        test_demo(os.path.join(directory, "agent_runs.sqlite3"))
    print("PASS: restart, checkpoint, event dedup, idempotency, terminal guard")

10.2 示例验证了什么，没有验证什么

已覆盖的机制：SQLite 持久化、原子状态与事件提交、递增版本、检查点、重复事件不重复生效、重启后读取状态、终态保护、同一操作 ID 的本地去重。由于示例使用 SQLite 单进程顺序测试，version 的 CAS 条件虽然写入了 SQL，却没有通过两个独立进程的竞争测试验证并发调度；真实系统需要额外的多连接或多进程竞争测试。

尚未覆盖的机制：真实外部 API 发送成功但响应丢失、Outbox Dispatcher、审批主体认证、操作参数哈希绑定、真实模型与工具调用、Worker 租约、跨版本迁移、日志脱敏、生产级数据加密和负载测试。尤其是 publish_once() 在本地数据库中模拟发布成功，不能将其视为跨网络外部发布的幂等性证明。

10.3 如何将原型推进到系统级

下一阶段应增加 attempts 表、operation_intents 表、审批记录、Outbox 与 Dispatcher；将发布模拟器替换成支持查询和幂等键的测试 HTTP 服务；使用两个独立进程争抢同一 Run，验证版本冲突；在请求发送前、远端执行后和本地提交前分别注入进程崩溃。最后加入实际模型调用，但不要让模型决定权威状态转换。

十一、验证矩阵、故障注入与质量指标

测试场景	注入方式	预期不变量	观测指标
正常完成	标准任务	合法状态序列、验收通过	完成率、延迟
输入非法	缺少必填字段	未调用工具	接纳失败率
双 Worker 竞争	同时更新相同 version	只有一次转移成功	冲突次数
事件重复	相同 event_id 投递两次	状态不重复推进	去重命中数
审批等待后重启	进程关闭并重新打开	等待与游标保留	恢复成功率
审批被撤销	恢复前改变授权	禁止敏感动作	越权拦截率
远端成功、响应丢失	工具模拟超时	不重复产生业务效果	副作用重复率
取消与完成竞争	并发取消/结果事件	状态与外部事实一致	取消竞态次数
预算耗尽	达到步骤上限	不再启动新行动	超预算执行数
Release 迁移	改变状态 Schema	拒绝不兼容恢复	迁移失败率
终态重启	COMPLETED 后恢复	不重新执行工具	非法重启次数

测试应分层：确定性状态机单元测试、SQLite/HTTP 模拟集成测试、崩溃与竞争故障注入、真实模型端到端任务质量测试。状态机测试通过不能证明任务研究质量合格；任务成功率高也不能证明重试、权限或取消机制可靠。

生命周期指标至少分成四组：状态指标（各状态停留时间、等待原因、超时率）；恢复指标（恢复成功率、平均恢复时长、恢复后重复执行数）；副作用指标（未知状态比例、核对成功率、重复业务效果）；版本与质量指标（Release 回归率、迁移成功率、业务验收通过率）。关联键应包含 run_id、release_id、step_id、operation_id、approval_id，并遵循敏感信息脱敏原则。

十二、工程取舍、常见错误与领域边界

12.1 常见错误与直接后果

设计错误	直接后果	核心修正
Session 与 Run 混用	多任务互相污染	独立 Run 身份、预算与审批
Run 与 Worker 进程绑定	进程崩溃即丢失任务	持久化 Run 与可接管执行权
只保存模型对话	无法判断外部动作是否发生	Checkpoint + Operation Ledger
模型自报完成即终止	产物缺失却标记成功	业务验收和产物核验
工具超时就直接重试	重复发送、重复写入	稳定操作 ID 与结果核对
等待状态依赖内存	重启后审批丢失	持久化等待事件与游标
旧审批可授权新参数	越权或发布错误内容	绑定参数哈希、期限与主体
修改 Release 覆盖历史	行为不可归因	不可变发布与显式迁移
取消被当作回滚	已发生副作用被忽略	记录事实并单独补偿

12.2 什么时候需要复杂生命周期架构

短时、只读、单用户的 Agent 可以从创建、运行、完成/失败、硬预算和基础日志起步。出现长时间等待时增加持久化 Checkpoint；出现外部写入时增加操作账本、幂等和审批；出现多用户与多 Worker 时增加租约、并发控制和租户隔离；出现频繁升级时增加不可变 Release 和迁移协议。复杂度应由真实故障模型和业务要求驱动，而不是为了凑齐所有模块。

12.3 与 Agent System / Agent Platform 的边界

本篇研究单 Agent 的生命周期语义与工程契约：哪些对象存在、何时允许转移、恢复需要什么事实、哪些外部动作不能重复。Agent System 进一步研究执行引擎、分布式调度、消息队列、持久化、故障恢复和观测基础设施；Agent Platform 进一步研究 Agent Registry、发布流程、实例控制、版本治理与组织级权限。单 Agent 同样可以依赖复杂系统基础设施，研究边界不是按 Agent 数量机械划分。

十三、核心结论

Agent 生命周期是多个对象和时间尺度叠加形成的系统：Definition 决定可复用能力，Release 固定可追溯配置，Instance 提供执行资源，Run 代表业务执行，Step 与 Operation 记录具体行动，Checkpoint 支持恢复，Session、Memory 与 Artifact 维持各自的数据寿命。生命周期管理的关键不是画出状态图，而是把合法转移、原子提交、执行事实、检查点、幂等操作、权限复核、版本兼容和终止验收落实为可验证的运行契约。

对工程实践最重要的三条原则是：模型输出不能代替权威状态；检查点不能代替外部副作用核对；重试不能代替幂等和业务一致性。 掌握这些机制，才能进一步设计长期可靠的 Agent Runtime，并在 Agent System 与 Agent Platform 中实现可运营、可追溯的任务执行能力。

参考资料与延伸阅读

1. Stuart Russell & Peter Norvig. Artificial Intelligence: A Modern Approach, 4th ed., 2021。智能体与环境的基础理论。
2. Shunyu Yao et al. ReAct: Synergizing Reasoning and Acting in Language Models, ICLR 2023。https://arxiv.org/abs/2210.03629 。Agent 行动循环研究。
3. Anthropic. Building Effective Agents, 2024。https://www.anthropic.com/engineering/building-effective-agents 。Agent 与工作流设计。
4. OpenAI. A Practical Guide to Building Agents, 2025。https://cdn.openai.com/business-guides-and-resources/a-practical-guide-to-building-agents.pdf 。工具、护栏与 Agent 工程实践。
5. LangGraph 官方文档：Persistence、Durable Execution、Interrupts、Human-in-the-loop。https://docs.langchain.com/oss/python/langgraph/overview 。图状态、检查点与恢复的框架资料。
6. Temporal 官方文档：Workflow Execution、Event History、Replay、Activity、Retry。https://docs.temporal.io/ 。持久化工作流与重放。
7. AWS Builders' Library. Making retries safe with idempotent APIs. https://aws.amazon.com/builders-library/making-retries-safe-with-idempotent-APIs/ 。幂等键与重试。
8. Martin Kleppmann. Designing Data-Intensive Applications, 2017。事务、事件与分布式一致性。
9. OpenTelemetry Documentation. Traces. https://opentelemetry.io/docs/concepts/signals/traces/ 。执行链路关联与追踪。
10. OWASP. Top 10 for LLM Applications. https://genai.owasp.org/ 。LLM 应用安全风险。

研究声明

来源与推导边界：本文以原 v1.0 的对象划分、运行状态、重试和数据生命周期研究为基础，结合通用状态机、事务、分布式一致性和代表性工作流框架的稳定技术概念进行机制深化。状态名称、SQLite 表结构、恢复算法和工具操作协议是本文的参考工程设计，不是 LangGraph、Temporal 或任何厂商的标准实现。

最新性说明：本次修订未进行逐条在线资料核验，不能声称已确认 2026 年 10 月 28 日最新 LangGraph/Temporal SDK、版本变更、源码细节或 API 兼容性。引用网址是后续验证入口，不代表本次逐页检索。真实选型和部署时应按实际版本重新核对官方文档。

代码与实验说明：本文提供可使用 Python 3 标准库运行的 SQLite 离线原型；其验证结果仅涉及本地事务、状态、检查点和模拟副作用，不代表真实模型、网络服务或分布式故障场景已通过测试。是否已实际执行本地测试，以本次交付的测试记录为准。