报告已经写进业务库Agent 却还没来得及保存“节点完成”进程就退出了。重启后检查点仍是“未完成”同一个工具于是又执行一遍。这次实验先复现出两条相同回执再把顺序倒过来先保存完成检查点后执行业务。重复消失了回执却变成零条。问题落在两次独立提交之间换顺序只会换一种出错方式。用两个 SQLite 文件分别保存流程状态和工具回执Python 子进程在指定位置直接退出再由另一个进程恢复。整个程序独立运行数据库里留下的记录就是要观察的结果。先看退出以后留下了什么每个用例只执行同一个逻辑动作保存demo-report-17。done1表示流程认为这个动作已经完成回执条数表示业务库实际接受了几次写入。首次执行在指定位置以退出码91结束随后换一个新进程恢复。2026-11-01Windows 11、Python 3.12.14、SQLite 3.53.1 的三组独立运行结果一致写法与退出位置退出后done / 回执数恢复后done / 回执数先提交业务写入再保存检查点中间退出0 / 11 / 2先保存完成检查点再提交业务写入中间退出1 / 01 / 0两项放在同一个本地事务提交前退出0 / 01 / 1两项放在同一个本地事务提交后退出1 / 11 / 1接收端按稳定操作键去重业务提交后、检查点前退出0 / 11 / 1第一行的0 / 1最关键流程不知道这次调用已经生效业务库却已经接受了它。恢复逻辑看到done0只能再次进入这个步骤。第二次提交成功后回执数从 1 变成 2。第二行把done1提前写入。恢复进程因此跳过步骤而退出前尚未执行的业务也再没有机会补上。流程记录宣布成功业务端却没有记录这是提前保存完成状态的代价。图1说明关注业务提交和检查点提交之间的窗口恢复读取的是持久状态无法从缺失的完成记录反推出业务端有没有成功。复现这五种退出位置把下面保存为checkpoint_demo.py使用 Python 3.12 或更高版本运行。只依赖标准库不访问外部服务。它默认创建一个新的临时目录保留数据库和results.json方便检查也可以传入一个尚不存在的输出目录。python-X utf8 checkpoint_demo.py# 或者保留到指定的新目录目录已存在时会直接报错python-X utf8 checkpoint_demo.py.\checkpoint-run-01-X utf8统一输出编码。最后一行应包含assertions: passed。外层程序会检查首次 worker 以预定的91退出、恢复 worker 返回0并核对两次数据库状态。任一检查不符或等待超时程序就报错。Finite process-crash fixture. Only writes to a newly created output directory.importjsonimportosfrompathlibimportPathimportplatformimportsqlite3importsubprocessimportsysimporttempfile KEYjob-17:save-report:v1PAYLOADdemo-report-17defconnect(path):dbsqlite3.connect(path,isolation_levelNone)db.execute(PRAGMA synchronousFULL)returndbdefinit(path,uniqueFalse):dbconnect(path)db.executescript( CREATE TABLE progress(id INTEGER PRIMARY KEY, done INTEGER NOT NULL); INSERT INTO progress VALUES(1, 0); CREATE TABLE receipts(op_key TEXT NOT NULL, payload TEXT NOT NULL); )ifunique:db.execute(CREATE UNIQUE INDEX receipt_key ON receipts(op_key))db.close()defemit(**record):print(json.dumps(record,ensure_asciiFalse),flushTrue)defcrash(point,inject):ifpointinject:emit(pidos.getpid(),eventinjected_exit,pointpoint)os._exit(91)# Deliberate process exit; no finally/close handler runs.defworker(root,mode,inject):cpconnect(root/control.db)effectcpifmodeatomicelseconnect(root/effect.db)ifcp.execute(SELECT done FROM progress WHERE id1).fetchone()[0]:emit(pidos.getpid(),eventcheckpoint_skip)else:definsert():effect.execute(INSERT INTO receipts VALUES(?, ?),(KEY,PAYLOAD))defcomplete():cp.execute(UPDATE progress SET done1 WHERE id1)ifmodecheckpoint_first:complete()crash(checkpoint,inject)insert()elifmodeatomic:cp.execute(BEGIN IMMEDIATE)insert()crash(in_transaction,inject)complete()cp.execute(COMMIT)crash(committed,inject)elifmodeidempotent:effect.execute(BEGIN IMMEDIATE)effect.execute(INSERT INTO receipts VALUES(?, ?) ON CONFLICT(op_key) DO NOTHING,(KEY,PAYLOAD))savedeffect.execute(SELECT payload FROM receipts WHERE op_key?,(KEY,)).fetchone()[0]ifsaved!PAYLOAD:effect.execute(ROLLBACK)raiseValueError(same operation key with a different payload)effect.execute(COMMIT)crash(effect,inject)complete()else:insert()crash(effect,inject)complete()emit(pidos.getpid(),eventstep_done)ifeffectisnotcp:effect.close()cp.close()defsnapshot(root,mode):cpconnect(root/control.db)effectcpifmodeatomicelseconnect(root/effect.db)result{done:cp.execute(SELECT done FROM progress WHERE id1).fetchone()[0],receipts:effect.execute(SELECT op_key,payload FROM receipts).fetchall(),}ifeffectisnotcp:effect.close()cp.close()returnresultdefrun_suite(output):# Distinct directories preserve each cases actual files and output.cases[(effect_first,effect,0,1,1,2),(checkpoint_first,checkpoint,1,0,1,0),(atomic,in_transaction,0,0,1,1),(atomic,committed,1,1,1,1),(idempotent,effect,0,1,1,1),]records[]formode,point,bd,br,ad,arincases:rootoutput/f{mode}-{point}root.mkdir()init(root/control.db)ifmode!atomic:init(root/effect.db,unique(modeidempotent))base[sys.executable,-X,utf8,__file__,--worker,str(root),mode]firstsubprocess.run(base[point],capture_outputTrue,textTrue,encodingutf-8,timeout10)beforesnapshot(root,mode)replaysubprocess.run(base[none],capture_outputTrue,textTrue,encodingutf-8,timeout10)aftersnapshot(root,mode)assertfirst.returncode91,first.stderrassertreplay.returncode0,replay.stderrassert(before[done],len(before[receipts]))(bd,br)assert(after[done],len(after[receipts]))(ad,ar)record{case:root.name,first_exit:first.returncode,first_stdout:first.stdout,first_stderr:first.stderr,before_replay:before,replay_exit:replay.returncode,replay_stdout:replay.stdout,replay_stderr:replay.stderr,after_replay:after}records.append(record)emit(caseroot.name,before[before[done],len(before[receipts])],after[after[done],len(after[receipts])],first_exitfirst.returncode,replay_exitreplay.returncode)report{python:platform.python_version(),sqlite:sqlite3.sqlite_version,platform:platform.platform(),cases:records,assertions:passed}(output/results.json).write_text(json.dumps(report,ensure_asciiFalse,indent2),encodingutf-8)emit(assertionspassed,outputstr(output))if__name____main__:iflen(sys.argv)1andsys.argv[1]--worker:worker(Path(sys.argv[2]),sys.argv[3],sys.argv[4])else:iflen(sys.argv)1:outputPath(sys.argv[1]).resolve()output.mkdir(parentsTrue,exist_okFalse)else:outputPath(tempfile.mkdtemp(prefixcheckpoint-replay-))run_suite(output)第一行输出应包含{case: effect_first-effect, before: [0, 1], after: [1, 2], first_exit: 91, replay_exit: 0}results.json保存两个 worker 的输出、进程 ID 和恢复前后重新查询的完整回执记录。os._exit(91)会跳过正常的finally和连接关闭逻辑后面的观察来自新连接读取的是退出后留下的数据库状态。这里模拟的是表中指定位置的进程退出。操作系统崩溃、断电、磁盘损坏和网络超时还涉及其他故障条件SQLite 的原子提交文档说明了其机制与前提。同一个事务能修复到哪里atomic分支把业务回执和done放进同一个 SQLite 文件、同一个显式事务。提交前退出恢复时两项都是旧值提交后退出两项都是新值。于是恢复进程不会遇到第一行那种“业务成功流程没记住”的分裂状态。两次写入必须真的共享这个事务。BEGIN IMMEDIATE开始本地写事务COMMIT提交两项修改具体语义见 SQLite 事务说明。HTTP 请求、邮件发送和另一个数据库的提交仍然在它的控制范围之外。如果业务动作本来就是本地数据更新把“处理记录”和业务数据一起提交通常更直接。不要为了套 Agent 的流程模型反而把原本可以原子提交的两次写入拆开。这个示例顺序恢复同一任务没有验证多个 worker 同时抢占任务的情形。实际有并发执行者时还需要在同一个事务内完成任务认领或状态条件更新不能只依赖事务外的一次SELECT done。工具在另一个系统时让重试有稳定身份idempotent分支保留两个独立数据库。业务端用op_key的唯一索引约束同一个逻辑动作只接受一份回执。第一次业务提交后进程退出恢复再次携带相同的键接收端读到已有记录再让流程保存完成状态。因此表里的回执数保持为 1。键写成job-17:save-report:v1意图是把它绑定到一次具体业务操作。它需要在重试前稳定下来并在恢复时继续使用。每重试一次就生成新 UUID会让每次调用看起来都是新的动作去重也就失效了。反过来复用键却偷偷改变参数同样危险。代码在冲突后还会比较payload同键不同内容直接报错。实际接口往往还要保存原结果让重试拿到同一个业务结果而不只是一个“已经存在”的提示。唯一约束提供了接收端去重基础保留多久、参数怎样绑定、失败结果能否重试都属于接口契约。唯一索引保证的是这张表只保留一份业务记录。接收端若还要调用不支持幂等的外部接口那个调用前后会重新出现同样的窗口。因此最终一条回执可能对应多次请求不能据此认定跨系统恰好执行一次。图2说明比较上表五种写法的实际持久结果。两种修复各自依赖不同条件同事务需要共享提交边界稳定操作键需要接收端遵守去重契约。接到 Agent 框架时先找到真实提交边界LangGraph 当前检查点文档把检查点描述为流程状态快照也明确说明从旧检查点重放时后续节点及其中的模型、API 调用会重新执行。接入工具时需要分别确认节点状态何时保存、内部业务动作何时生效。这里引用官方说明来定位接入边界上面的运行数据来自独立 Python 程序。落到一个具体工具先回答三个问题业务成功由哪一端确认这个结果什么时候成为持久状态如果响应丢失恢复进程用什么身份查询或重试。读操作一般可以直接重试但“读取即消费”、生成新资源、发送通知都要按真实副作用分析不能只看函数名。如果接收端既不能查询结果也没有幂等契约恢复后应该保留“结果未知”并进入核对流程。把未知直接改成失败再自动重试可能重复操作把未知直接改成成功又可能遗漏操作。检查点里保存这个未决状态比保存一个没有证据的完成标记更有用。
