一、断线以后，最危险的不是任务没跑完，而是结果不确定

Agent 执行任务时，网络断开、浏览器标签崩溃、模型请求超时、电脑休眠，都可能发生在一个很尴尬的时间点：外部动作已经成功，执行端却没来得及收到确认。恢复后，系统只知道“上次调用超时”，不知道那一步究竟有没有生效。

比如一个 Agent 收到“给客户生成一次报告并通知负责人”的任务。它先把报告写入业务系统，随后调用通知接口。通知服务已经接收，但响应包在回程中丢失。调度器看到超时，就把整个任务再跑一遍。第二次报告可能重复创建，通知也可能发两遍。把超时简单理解为失败，是这类事故的起点。

这和普通脚本报错不一样。脚本停在本地计算阶段，重跑通常只是多花一点时间；一旦中间包含发邮件、扣款、创建工单、发布文章、提交代码等外部副作用，重跑就会改变真实世界。恢复机制要回答的不是“从哪一行继续”，而是“哪些结果已经可靠提交，哪些动作可以安全重试”。

本文用 SQLite 做一个小型任务账本，模拟两处最常见的断线窗口：本地事务提交前进程退出，以及下游已经接收消息、上游却没收到确认。实验不访问公网，也不调用真实支付、邮件或发布接口。我们只用两份本地数据库，验证任务键、事务、Outbox 和接收端去重各自解决什么问题。

二、先拆开三个状态：计算完成、结果提交、外部通知

很多任务系统只保存一个 status 字段，例如 pending、running、done。字段看起来简单，遇到断线却很难判断。因为“函数执行到最后一行”“业务数据已经写入”“外部系统已经接收”是三个不同事实，不能共用一个完成标记。

第一层是计算状态。模型已经生成结果，脚本也得到了输出，但这份输出可能只在内存里。进程被结束，结果就消失了。第二层是本地提交状态。结果和任务检查点在同一个数据库事务里提交，重启后可以从数据库读回来。第三层是外部交付状态。邮件服务、工单系统或另一台机器是否收到事件，属于另一个系统的事实，不能靠本地变量推断。

因此，任务账本至少要保存稳定的 task_key、输入指纹、阶段状态和结果引用；如果本地提交后还要通知外部系统，还需要一张待发送事件表。接收端也应记录自己处理过的 event_id。这样，恢复时才有证据判断是继续计算、读取已提交结果，还是重试一条可以去重的通知。

这套设计不是为了把每个任务都变成复杂工作流。只有当任务包含不可逆动作、跨系统通知或长时间运行时，才值得把这些状态明确保存。纯本地、可随时重算、没有外部副作用的短脚本，通常不需要这么多表。

三、为什么只保存 last_step 仍然会重复

最直观的方案是在每一步结束后写入 last_step=3。恢复时读取这个值，从第四步继续。它比完全没有状态好，但无法覆盖“动作已经生效、检查点尚未写入”的窗口。

假设第三步调用外部接口创建工单。接口返回成功后，程序准备把 last_step 更新为 3；就在两者之间进程崩溃。重启后数据库仍显示 last_step=2，程序会再次调用创建工单。相反，如果先把 last_step 写成 3，再调用外部接口，程序可能在写检查点后、调用接口前崩溃，恢复时就会跳过一个实际上从未发生的动作。

把顺序反过来并不能消除问题，只是把重复执行换成漏执行。原因是本地数据库和远端 HTTP 服务没有共享一个原子事务。除非远端接口支持幂等键，或双方参与同一个可靠事务协议，否则本地无法单方面证明远端动作只发生了一次。

所以，检查点要和本地业务结果放在同一事务里；跨系统动作则用 Outbox 保存“待发送的事实”，交给独立发送器重试，并要求接收端按事件 ID 去重。这里的目标不是宣称网络只传一次，而是让同一事件传多次也不会产生多份业务结果。

四、SQLite 账本的四张表分别负责什么

实验使用四类记录。tasks 保存任务键、输入指纹、状态和结果；reports 模拟任务真正生成的本地业务结果；outbox 保存需要发往下游的事件；receiver_inbox 保存下游已经接收过的事件 ID。它们并不是四种队列，而是分别记录不同系统边界上的事实。

task_key 必须由业务输入稳定生成，不能每次重试都随机创建。比如同一个客户、同一个报告周期、同一份输入版本，可以组合成一个任务键。输入指纹则用于防止调用方错误复用键：如果相同 task_key 对应的 payload_hash 不同，程序应该明确报冲突，而不是悄悄把新输入当成旧结果。

reports 和 outbox 必须在同一个 SQLite 事务里写入。这样，事务提交后，任务结果和待发送事件同时存在；事务回滚时，两者同时不存在。若只写了报告、没有写 outbox，外部通知可能永远不会发出；若只发通知、没有保存报告，接收方可能看到一个本地并不存在的结果。

接收端 inbox 的 event_id 设置为主键。示例先在同一事务中查询 event_id，不存在才插入；相同事件再次到达时，会识别为已存在并跳过新增记录。主键约束负责挡住重复写入。真实业务中，这个去重记录通常应当落在接收方自己的数据库，并与接收方业务写入处于同一事务；示例把它单独放在 receiver.db，模拟两个系统之间的边界。

五、事务边界：只把本地能一起提交的动作放进去

第一段代码刻意把“计算”和“提交”分开。昂贵的模型调用、文件解析、HTTP 查询，不应该在持有 SQLite 写锁时执行。程序先在事务外准备好 payload，再开启短事务，检查 task_key，写入任务结果、报告记录和 outbox 事件，最后提交。

这里使用 BEGIN IMMEDIATE。它会在事务开始时尝试取得写事务，而不是等到执行写语句时才升级。这样两个恢复进程同时抢同一个任务时，竞争会在事务入口显现；第二个写入者可能等待，也可能在超时后收到 SQLITE_BUSY。它不是分布式锁，也不会让 SQLite 同时拥有多个写入者。数据库唯一约束仍然是最终防线。

WAL 模式可以让读者和写者更好地并行，但不能把 SQLite 变成多写者数据库。官方文档明确说明，WAL 下仍然只有一个 writer；所有访问同一 WAL 数据库的进程还必须位于同一主机，不能把 WAL 文件放在网络文件系统上供多台机器共享。

下面代码的两个数据库都位于临时目录。第一段模拟任务结果提交；第二段模拟 Outbox 投递和接收端幂等。脚本会打印每个断线窗口后的记录数，并用 assert 检查没有多出业务结果。

下面的流程把本地提交和外部投递分成两段。任务结果与 outbox 同时提交；发送器随后投递同一个 event_id；接收端先查 inbox，再决定是否执行副作用。

Agent / Worker
  └─ BEGIN IMMEDIATE
       ├─ tasks + reports + outbox
       └─ COMMIT
             ↓
        Outbox Sender
             ↓ event_id
        Receiver Inbox
             ├─ 新事件：执行业务并提交
             └─ 已存在：返回原结果

还要区分任务身份和执行尝试。一次任务可以经历多次 attempt，但它们仍属于同一个 task_key。不要把 attempt_id 当成幂等键：每次重试都生成新的 attempt_id，接收端就无法识别它们其实是同一业务意图。可以同时保存 attempt_id 用于诊断，把 task_key 或 event_id 用于去重；两者解决的问题不同。

唯一键也需要有明确的保留策略。若系统只保留最近七天的任务记录，而某个延迟队列在第八天才重投，旧事件就可能失去去重依据。保留期至少要覆盖最大重试窗口、消息滞留时间和人工补偿周期。对账、删除和归档不能随意清掉仍可能被重放的键，否则幂等保护会在最需要的时候失效。

六、完整实验：提交前崩溃、提交后丢确认

先看第一种情况。程序开始一个事务，写入 task、report 和 outbox，然后在 commit 之前主动回滚，模拟进程在提交前退出。重新打开数据库后，三张表都应该是零条记录。接着正常提交一次，再把同一个 task_key 重放。第二次调用必须返回 already_committed，而不是再插入一份报告。

这里的“崩溃”由脚本主动 rollback 模拟，不能等同于拔电源或杀进程的完整故障注入。它验证的是事务边界：提交前的写入没有成为可见的已提交结果。SQLite 自己负责事务恢复；应用负责使用稳定键和唯一约束识别重放。

第二种情况更容易被忽略。发送器读取 outbox 后，把 event_id 发给 receiver。receiver 已经把事件写入自己的 inbox 并提交，但发送器在更新 outbox 状态前崩溃。此时发送器只能看到 pending，下一轮会再次发送。接收端发现 event_id 已存在，就返回 duplicate，不再重复写入业务结果；发送器随后把 outbox 标记为 delivered。

这就是至少一次投递加幂等消费。消息可能被重复传输，但业务处理结果只有一份。代码没有把“HTTP 200”当成最终证据，也没有假设网络调用只会发生一次。

七、先明确实验步骤：脚本会模拟两个断线窗口

本节先看断点和预期结果，再在文末附录运行完整程序。脚本只依赖 Python 标准库，在临时目录创建 sender.db 和 receiver.db，打印 Python、SQLite 版本及每个阶段的记录数量；临时目录在程序结束后自动清理，不会修改电脑上的其他文件。

八、如何读运行结果：四个断点，四种结论

运行后，第一行应该显示 pre_commit_crash 后 tasks、reports、outbox 都为 0。这说明本地事务在提交前被回滚，恢复进程不会把半套写入误认成完成。第二行 first_commit 应为 committed；第三行 lost_ack_retry 应为 already_committed，任务和报告数量仍然都是 1。

接下来，receiver_commit_before_sender_ack 模拟接收端已保存事件、发送端还没更新状态。此时 sender 的 outbox 仍是 pending，而 receiver 的 inbox 已有 1 条。重试发送后，脚本返回 duplicate_deduplicated，说明同一个 event_id 已经存在并被识别为重复；最终 outbox 状态变为 delivered，receiver inbox 仍然只有 1 条。

这组实验验证的是幂等边界，不是吞吐性能。没有对并发数、磁盘速度或远端网络做压测，也没有声称一次本地模拟能证明所有故障都不会丢数据。它能证明的是：本地任务结果和待发送事件可以原子提交；任务重放可以通过稳定键识别；接收端可以用唯一事件 ID 抵御重复投递。

如果你看到任务表只有一条，却发现远端服务创建了两张工单，说明远端没有执行同一幂等合同，或者本地 task_key 与远端幂等键不是同一个业务标识。此时不能只在本地加一张表就宣称问题解决。

九、为什么 WAL、检查点和 Outbox 不是一回事

这三个词经常一起出现，却解决不同层面的问题。SQLite WAL 是数据库的日志模式：提交时把变更追加到 WAL，读者可以继续读取稳定快照，之后再 checkpoint 把内容回写主数据库。它解决的是数据库读写并发和提交机制，不负责判断某个 Agent 任务是否重复。

任务 checkpoint 是应用层的恢复记录。它告诉调度器某个 task_key 的结果已经提交到哪里，以及恢复时应该读取哪个结果。它必须和相应业务数据处于同一事务，否则仍会出现“数据已写但状态没写”或“状态已写但数据没写”的窗口。

Outbox 是跨系统交付的持久化缓冲。业务事务把事件写入 outbox，发送器异步投递；投递失败可以重试，成功后再标记 delivered。它不保证消息只发一次，也不保证接收方只执行一次。接收方还需要自己的 inbox 或唯一业务键，把重复消息转换为同一个结果。

因此，启用 WAL 不等于有任务恢复；有 task 表不等于外部通知可靠；有 outbox 也不等于接收端不会重复。把三者混成一个“开了 SQLite 就安全”的结论，反而会掩盖真正的系统边界。

十、把实验接进 Agent：恢复时应该先查证据

实际 Agent 执行器可以把任务流程收敛为几个清晰动作：接收请求时生成稳定 task_key；执行前计算 payload_hash；恢复时先查询 task 表；若已提交且输入指纹一致，直接返回保存的结果；若键相同但指纹不同，拒绝覆盖；若没有记录，才进入计算阶段。计算完成后，用一个短事务写入结果和 outbox。

发送器独立轮询 pending 事件，按 event_id 投递。遇到超时，不把它当作“远端一定失败”，而是保持事件待确认并安全重试。接收端把 event_id 与业务结果一起提交；如果该事件已经处理过，就返回之前的结果或明确的重复标记。这样，发送器即使在远端成功后崩溃，也不需要猜测是否应当再执行一遍。

恢复程序还应保存输入摘要、输出摘要、执行版本和阶段时间。只保存 task_key 可能不够：同一任务在代码升级后，计算规则也可能改变。可以把 worker_version、prompt_version、schema_version 纳入 payload_hash 或任务元数据，明确区分“同一任务重试”和“新版本重新计算”。不要把模型生成的自然语言标题当成唯一键，标题可能被改写或重复。

对浏览器自动化也一样。点击发布按钮之前可以保存准备状态，但真正的发布结果必须回到平台后台查证；平台已发布但本地没收到响应时，应先按文章标题、账号和时间范围查找原结果，不能直接再次点击。数据库只能帮助记录自己的动作，无法替代对外部平台终态的读取。

十一、上线前的决策表：哪些动作可以重试

任务类型不同，恢复策略也不同。纯计算、可重复下载、只读查询通常可以直接重试；本地数据库写入应使用事务和唯一键；发邮件、创建工单、发布内容等外部动作，优先使用对方支持的幂等键；如果没有幂等接口，就要先查证状态，或设计人工确认与补偿流程。

场景	恢复依据	推荐动作
计算中断，尚无提交记录	task_key 不存在	重新计算
本地事务提交前退出	事务回滚，记录不存在	重新执行短事务
本地提交后响应丢失	task_key 和 payload_hash 一致	返回已保存结果
Outbox 投递超时	event_id 仍待确认	重试同一个 event_id
接收端已处理，确认丢失	receiver inbox 已有 event_id	返回已有结果，不重复副作用
同一 task_key 输入发生变化	payload_hash 不一致	拒绝重放，创建新版本任务

这张表最重要的一行，是“提交后响应丢失”。程序没有收到确认，不代表数据库没有提交；恢复时必须先读状态，而不是重新做副作用。相反，如果同一个 task_key 对应了不同输入，直接复用旧结果同样是错误的。幂等不是无条件跳过，而是“同一业务意图、同一输入版本，重复请求得到同一结果”。

如果下游接口支持 Idempotency-Key，建议把稳定 event_id 传给它，并确认该接口的键作用域、保留时间和冲突响应。键过期后再重试，可能又创建一次新对象；不同租户共用同一键，也可能产生错误去重。没有这些合同，不能仅凭请求头里出现一个 Idempotency-Key 就认定远端真的支持幂等。

十二、什么时候 SQLite 不再合适

SQLite 很适合单机 Agent、桌面自动化、轻量任务队列和本地持久化检查点。它部署简单、事务能力成熟、没有额外数据库服务，尤其适合把运行状态保存在执行机器上。但它不是多机分布式协调服务，也不是无限吞吐的消息队列。

如果多个主机需要同时抢任务，或者写入量高到一个写者成为瓶颈，应考虑 PostgreSQL 等服务端数据库或专用队列。若任务执行跨越多个服务，并且每个服务都有独立数据存储，应该设计 Outbox、Inbox、幂等消费者或 Saga，而不是试图把多个服务的数据库文件放到共享目录里。SQLite WAL 官方明确不支持跨主机网络文件系统的共享场景。

__BODY_IMG_PLACEHOLDER__

还要考虑数据库备份和恢复。WAL 模式下，主数据库旁边可能存在 -wal 和 -shm 文件；备份不能只在数据库正在写入时随手复制一个主文件，就宣称拿到了完整一致的快照。应使用 SQLite 提供的备份接口或受支持的备份流程，并定期验证恢复。长时间读事务也可能阻碍 checkpoint 完成，让 WAL 文件持续增长。

最后，SQLite 的事务原子性不是对所有硬件、文件系统和外部设备的绝对承诺。数据库文件应放在本机可靠文件系统，保持合适的 synchronous 设置，避免让其他程序绕过 SQLite 直接改数据库文件。真正重要的任务还要有备份、监控、磁盘空间告警和故障演练。

十三、完整可运行的 Python 实验代码

将下面代码保存为 checkpoint_outbox_demo.py，使用 Python 3.11 及以上版本运行。它包含本地事务、任务键冲突、Outbox 投递和接收端去重的完整演示。

from pathlib import Path
import hashlib
import json
import sqlite3
import tempfile


class SimulatedCrash(RuntimeError):
    pass


def connect(path):
    db = sqlite3.connect(path, timeout=5)
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA synchronous=FULL")
    return db


def init_sender(path):
    db = connect(path)
    db.executescript("""
    CREATE TABLE tasks (
        task_key TEXT PRIMARY KEY,
        payload_hash TEXT NOT NULL,
        status TEXT NOT NULL,
        result TEXT NOT NULL
    );
    CREATE TABLE reports (
        report_key TEXT PRIMARY KEY,
        body TEXT NOT NULL
    );
    CREATE TABLE outbox (
        event_id TEXT PRIMARY KEY,
        task_key TEXT NOT NULL UNIQUE,
        payload TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending',
        attempts INTEGER NOT NULL DEFAULT 0
    );
    """)
    db.close()


def init_receiver(path):
    db = connect(path)
    db.execute("""
        CREATE TABLE inbox (
            event_id TEXT PRIMARY KEY,
            payload TEXT NOT NULL
        )
    """)
    db.close()


def commit_task(sender_path, task_key, payload, crash_before_commit=False):
    payload_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    result = {"task_key": task_key, "summary": "report:" + payload}
    result_json = json.dumps(result, ensure_ascii=False)
    event_id = task_key + ":completed:v1"

    db = connect(sender_path)
    db.execute("BEGIN IMMEDIATE")
    try:
        old = db.execute(
            "SELECT payload_hash, status FROM tasks WHERE task_key=?",
            (task_key,)
        ).fetchone()

        if old is not None:
            if old[0] != payload_hash:
                raise ValueError("同一 task_key 对应了不同输入")
            db.rollback()
            return "already_committed" if old[1] == "committed" else "already_exists"

        db.execute(
            "INSERT INTO tasks(task_key, payload_hash, status, result) VALUES(?,?,?,?)",
            (task_key, payload_hash, "committed", result_json)
        )
        db.execute(
            "INSERT INTO reports(report_key, body) VALUES(?,?)",
            (task_key, result_json)
        )
        db.execute(
            "INSERT INTO outbox(event_id, task_key, payload) VALUES(?,?,?)",
            (event_id, task_key, result_json)
        )

        if crash_before_commit:
            db.rollback()
            raise SimulatedCrash("模拟在本地 COMMIT 前退出")

        db.commit()
        return "committed"
    except Exception:
        if db.in_transaction:
            db.rollback()
        raise
    finally:
        db.close()


def deliver_one(sender_path, receiver_path, event_id, crash_after_receiver_commit=False):
    sender = connect(sender_path)
    row = sender.execute(
        "SELECT payload, status FROM outbox WHERE event_id=?",
        (event_id,)
    ).fetchone()

    if row is None:
        sender.close()
        raise KeyError(event_id)

    payload, status = row
    if status == "delivered":
        sender.close()
        return "already_delivered"

    receiver = connect(receiver_path)
    receiver.execute("BEGIN IMMEDIATE")
    existed = receiver.execute(
        "SELECT 1 FROM inbox WHERE event_id=?",
        (event_id,)
    ).fetchone() is not None

    if not existed:
        receiver.execute(
            "INSERT INTO inbox(event_id, payload) VALUES(?,?)",
            (event_id, payload)
        )

    receiver.commit()
    receiver.close()

    if crash_after_receiver_commit:
        sender.close()
        raise SimulatedCrash("模拟接收端已提交、发送端尚未确认")

    sender.execute("BEGIN IMMEDIATE")
    sender.execute(
        "UPDATE outbox SET status='delivered', attempts=attempts+1 WHERE event_id=?",
        (event_id,)
    )
    sender.commit()
    sender.close()
    return "duplicate_deduplicated" if existed else "delivered"


def counts(path, table):
    db = connect(path)
    value = db.execute("SELECT COUNT(*) FROM " + table).fetchone()[0]
    db.close()
    return value


def outbox_status(path, event_id):
    db = connect(path)
    row = db.execute(
        "SELECT status, attempts FROM outbox WHERE event_id=?",
        (event_id,)
    ).fetchone()
    db.close()
    return row


def main():
    print("python:", __import__("sys").version.split()[0])
    print("sqlite:", sqlite3.sqlite_version)

    with tempfile.TemporaryDirectory() as folder:
        sender = Path(folder) / "sender.db"
        receiver = Path(folder) / "receiver.db"
        init_sender(sender)
        init_receiver(receiver)

        task_key = "customer-42:report:2026-11"
        payload = "customer=42;period=2026-11"

        try:
            commit_task(sender, task_key, payload, crash_before_commit=True)
        except SimulatedCrash:
            pass

        print(
            "pre_commit_crash:",
            counts(sender, "tasks"),
            counts(sender, "reports"),
            counts(sender, "outbox")
        )

        first = commit_task(sender, task_key, payload)
        retry = commit_task(sender, task_key, payload)
        print("commit:", first, "lost_ack_retry:", retry)

        try:
            commit_task(sender, task_key, "different-input")
        except ValueError:
            print("same_key_different_payload: rejected")

        event_id = task_key + ":completed:v1"
        try:
            deliver_one(sender, receiver, event_id, crash_after_receiver_commit=True)
        except SimulatedCrash:
            pass

        print(
            "receiver_committed_before_sender_ack:",
            outbox_status(sender, event_id)[0],
            counts(receiver, "inbox")
        )

        redelivery = deliver_one(sender, receiver, event_id)
        print("redelivery:", redelivery)
        print(
            "final:",
            "tasks=", counts(sender, "tasks"),
            "reports=", counts(sender, "reports"),
            "outbox=", outbox_status(sender, event_id),
            "receiver_inbox=", counts(receiver, "inbox")
        )

        assert counts(sender, "tasks") == 1
        assert counts(sender, "reports") == 1
        assert counts(sender, "outbox") == 1
        assert counts(receiver, "inbox") == 1
        assert outbox_status(sender, event_id)[0] == "delivered"


if __name__ == "__main__":
    main()

十四、结论：不要追求“只执行一次”，要让重试没有副作用

Agent 任务断线后，最容易犯的错是把“没收到响应”当成“动作没发生”。如果动作已经越过系统边界，重跑就可能制造重复结果。更可靠的做法是把本地业务结果和检查点放在一个事务里，用稳定 task_key 和输入指纹识别重放；把跨系统通知写进 Outbox；让接收端用 event_id 或业务唯一键幂等处理。

本次本机实验验证了三个结果：事务提交前的写入可以整体回滚；本地提交后即使确认丢失，重复调用仍能读取原结果；接收端提交后发送端断线，重复投递也不会新增第二条接收记录。它没有证明所有远端系统都能做到 exactly-once，也没有替代真实接口的幂等合同。

如果只能先改一处，优先把“任务结果”和“完成检查点”放进同一个短事务，并给任务设置稳定唯一键。下一步再把外部动作迁到 Outbox，并要求接收端按同一个事件 ID 去重。这样做的目标不是让网络永不失败，而是让失败之后仍然知道发生过什么，并且能够安全地继续。

参考资料

Python 3.13 sqlite3 事务控制文档：https://docs.python.org/3.13/library/sqlite3.html

SQLite 事务控制：https://www.sqlite.org/lang_transaction.html

SQLite Write-Ahead Logging：https://www.sqlite.org/wal.html

AWS Prescriptive Guidance：Transactional outbox pattern：https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html