（十四）在失败重试时保证计划只保存一份

1. 推导可能重复的位置

假设 save 已提交 plans 表，但还没把完成结果写回图检查点，进程就退出。重开后，图只知道 save 尚未完成，于是会再次执行它。程序需要识别“同一项动作又来了一次”。

本项目在任务开始时生成 operation_id，存进图状态。计划表对 (owner, operation_id) 建唯一约束；再次保存同一个编号时读取原记录，并核对问题和时长是否一致。

# 对照完整文件阅读这一段；此处保留当前步骤使用的函数或类型。
class PlanStore:
    def __init__(self, path: Path):
        self.path = path
        path.parent.mkdir(parents=True, exist_ok=True)
        # 内层连接上下文处理事务，外层 closing 在结束时关闭连接。
        with closing(sqlite3.connect(path)) as connection, connection:
            connection.execute("""CREATE TABLE IF NOT EXISTS plans (
                id INTEGER PRIMARY KEY, owner TEXT NOT NULL, operation_id TEXT NOT NULL,
                question TEXT NOT NULL, minutes INTEGER NOT NULL,
                UNIQUE(owner, operation_id))""")

    def save(self, owner: str, operation_id: str, question: str, minutes: int) -> int:
        """相同业务编号的重复执行读取原记录，不再多插一行。"""
        with closing(sqlite3.connect(self.path)) as connection, connection:
            connection.execute("INSERT INTO plans(owner,operation_id,question,minutes) VALUES (?,?,?,?) "
                               "ON CONFLICT(owner,operation_id) DO NOTHING", (owner, operation_id, question, minutes))
            row = connection.execute("SELECT id,question,minutes FROM plans WHERE owner=? AND operation_id=?",
                                     (owner, operation_id)).fetchone()
            if row[1:] != (question, minutes):
                raise ValueError("同一个操作编号已经用于另一份计划")
            return row[0]

    def list(self, owner: str) -> list[dict]:
        with closing(sqlite3.connect(self.path)) as connection, connection:
            connection.row_factory = sqlite3.Row
            return [dict(row) for row in connection.execute("SELECT * FROM plans WHERE owner=? ORDER BY id", (owner,))]

连接管理沿用第五章的 contextlib.closing。同一条 with 中先进入 closing，再进入连接的事务上下文；退出时顺序反过来，先提交或回滚事务，再关闭连接。这样 save 返回 id 时，事务和连接都会完成各自的收尾。

ON CONFLICT ... DO NOTHING 处理已有编号；随后读取原行。若同一个编号却携带另一份内容，抛出错误而不是悄悄覆盖；同一内容则返回原 id。参数仍用占位符传入，延续第五章的 SQL 用法。

thread_id 和 operation_id 也要分清：前者用于找到这项图任务的进度，后者用于识别要保存的那一次业务动作。恢复失败节点时，这两个编号都沿用原值；若重新生成 operation_id，计划表就会把它当成另一项操作。

业务事务已提交但图节点失败时，使用原 operation_id 重试并读取原记录

2. 让测试真的经过失败位置

项目测试在业务提交后、节点返回前制造一次异常，再关闭检查点连接。重新打开连接后，以相同身份和任务号继续执行，最后核对：状态是 saved，计划只有一行，id 与第一次提交相同。

可运行的 crash_retry_walk.py 把这两份状态打印出来。失败之后，图状态还是 approved，next 指向 save，没有 pending，也没有 plan_id；计划表却已经有 id=1 的一行。这说明等待完成的是保存节点，用户的批准已经写回，无需再交付一次确认值。

重试使用 graph.invoke(None, ...)。save 拿到原来的 op-demo-1，唯一约束使第二次 INSERT 不再新增，再查回 id=1 并返回。最后图状态变成 saved，next 为空，计划表仍是同一行。这种重复执行后仍得到同一业务结果的处理，就是这里的幂等保存。

图的检查点与计划库各自提交，所以恢复时要同时观察“图知道什么”和“业务已经做了什么”。同样的做法也可以用于外部保存接口：沿用操作编号，查询原来的结果，再标记本次节点完成。