报告已经写进业务库，Agent 却还没来得及保存“节点完成”，进程就退出了。重启后，检查点仍是“未完成”，同一个工具于是又执行一遍。

这次实验先复现出两条相同回执，再把顺序倒过来：先保存完成检查点，后执行业务。重复消失了，回执却变成零条。问题落在两次独立提交之间，换顺序只会换一种出错方式。

用两个 SQLite 文件分别保存流程状态和工具回执，Python 子进程在指定位置直接退出，再由另一个进程恢复。整个程序独立运行，数据库里留下的记录就是要观察的结果。

先看退出以后留下了什么

每个用例只执行同一个逻辑动作：保存demo-report-17。done=1表示流程认为这个动作已经完成；回执条数表示业务库实际接受了几次写入。首次执行在指定位置以退出码91结束，随后换一个新进程恢复。

2026-11-01，Windows 11、Python 3.12.14、SQLite 3.53.1 的三组独立运行结果一致：

第一行的0 / 1最关键：流程不知道这次调用已经生效，业务库却已经接受了它。恢复逻辑看到done=0，只能再次进入这个步骤。第二次提交成功后，回执数从 1 变成 2。

第二行把done=1提前写入。恢复进程因此跳过步骤，而退出前尚未执行的业务也再没有机会补上。流程记录宣布成功，业务端却没有记录；这是提前保存完成状态的代价。

图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 KEY="job-17:save-report:v1"PAYLOAD="demo-report-17"defconnect(path):db=sqlite3.connect(path,isolation_level=None)db.execute("PRAGMA synchronous=FULL")returndbdefinit(path,unique=False):db=connect(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_ascii=False),flush=True)defcrash(point,inject):ifpoint==inject:emit(pid=os.getpid(),event="injected_exit",point=point)os._exit(91)# Deliberate process exit; no finally/close handler runs.defworker(root,mode,inject):cp=connect(root/"control.db")effect=cpifmode=="atomic"elseconnect(root/"effect.db")ifcp.execute("SELECT done FROM progress WHERE id=1").fetchone()[0]:emit(pid=os.getpid(),event="checkpoint_skip")else:definsert():effect.execute("INSERT INTO receipts VALUES(?, ?)",(KEY,PAYLOAD))defcomplete():cp.execute("UPDATE progress SET done=1 WHERE id=1")ifmode=="checkpoint_first":complete()crash("checkpoint",inject)insert()elifmode=="atomic":cp.execute("BEGIN IMMEDIATE")insert()crash("in_transaction",inject)complete()cp.execute("COMMIT")crash("committed",inject)elifmode=="idempotent":effect.execute("BEGIN IMMEDIATE")effect.execute("""INSERT INTO receipts VALUES(?, ?) ON CONFLICT(op_key) DO NOTHING""",(KEY,PAYLOAD))saved=effect.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(pid=os.getpid(),event="step_done")ifeffectisnotcp:effect.close()cp.close()defsnapshot(root,mode):cp=connect(root/"control.db")effect=cpifmode=="atomic"elseconnect(root/"effect.db")result={"done":cp.execute("SELECT done FROM progress WHERE id=1").fetchone()[0],"receipts":effect.execute("SELECT op_key,payload FROM receipts").fetchall(),}ifeffectisnotcp:effect.close()cp.close()returnresultdefrun_suite(output):# Distinct directories preserve each case's 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:root=output/f"{mode}-{point}"root.mkdir()init(root/"control.db")ifmode!="atomic":init(root/"effect.db",unique=(mode=="idempotent"))base=[sys.executable,"-X","utf8",__file__,"--worker",str(root),mode]first=subprocess.run(base+[point],capture_output=True,text=True,encoding="utf-8",timeout=10)before=snapshot(root,mode)replay=subprocess.run(base+["none"],capture_output=True,text=True,encoding="utf-8",timeout=10)after=snapshot(root,mode)assertfirst.returncode==91,first.stderrassertreplay.returncode==0,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(case=root.name,before=[before["done"],len(before["receipts"])],after=[after["done"],len(after["receipts"])],first_exit=first.returncode,replay_exit=replay.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_ascii=False,indent=2),encoding="utf-8")emit(assertions="passed",output=str(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:output=Path(sys.argv[1]).resolve()output.mkdir(parents=True,exist_ok=False)else:output=Path(tempfile.mkdtemp(prefix="checkpoint-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 程序。

落到一个具体工具，先回答三个问题：业务成功由哪一端确认；这个结果什么时候成为持久状态；如果响应丢失，恢复进程用什么身份查询或重试。读操作一般可以直接重试，但“读取即消费”、生成新资源、发送通知，都要按真实副作用分析，不能只看函数名。

如果接收端既不能查询结果，也没有幂等契约，恢复后应该保留“结果未知”并进入核对流程。把未知直接改成失败再自动重试，可能重复操作；把未知直接改成成功，又可能遗漏操作。检查点里保存这个未决状态，比保存一个没有证据的完成标记更有用。