长任务怎么恢复：从宕机现场到可运行 Harness

01 · 先看失败现场：为什么“重试一次”会生成两份报告

假设你要做一个公司内部技术研究助手。它读取两份获准材料，生成带来源 ID 的摘要，然后调用报告服务保存。任务可能运行数十分钟，调度消息可能重复投递，模型请求也可能超时。业务真正关心的是：已做完的工作不丢失，未知状态能解释，同一份报告不被重复写入。

这里的问题与模型是否聪明无关。即使输出完全确定，T1 到 T2 之间仍然有跨系统提交空隙。本地事务只能约束本地数据库，不能把另一个服务已经发生的写入回滚。解决时必须分别回答两件事：运行时从哪里恢复；业务写入重复到达时如何处理。

本篇结论： 用检查点确定下一步，用租约代次拒绝旧执行者写回，用稳定操作键和远端持久回执处理重试。三个机制各自承担不同职责，缺一个就有对应的故障窗口。

02 · 一个可验收的案例：四步、两个数据库、明确交付物

本次统一案例是固定四步研究流程。它用于验证 Agent 的执行外壳（Harness）：调度、状态、工具结果和恢复边界。默认 draft 使用确定性替身，便于复现；可选切换 HTTP 模型。它没有让模型动态规划工具路线，因此不能用这套结果证明自主 Agent 的规划能力。

运行后得到 lab.sqlite（任务、步骤、事件和记忆）及 lab.sqlite.publisher.sqlite（模拟外部报告服务）。刻意用两个独立数据库，让远端写入与本地检查点无法共享事务。网页展示的恢复结果就是从这两个库实际读取，材料内容是合成样例。

同一套案例，三篇文章共用

Python 3.10+ · 标准库 · 默认离线 · 含源代码、36 项测试和运行记录

下载完整实验包 ZIP运行说明

1. 新建空目录，解压下载包；进入包含 cli.py 的目录。
2. 执行下面命令。默认无需 pip install、账户或 API Key。
3. 用 inspect 检查 4 个检查点，再用 evaluate.py 查看机械验收结果。

第一轮：正常路径

python3 cli.py memory-put
python3 cli.py submit
python3 cli.py run
python3 cli.py inspect
python3 evaluate.py
python3 -m unittest discover -s . -p test_lab.py -v

预期 run 输出 succeeded。评分 JSON 中 passed 为 true，而 semantic_support 与 model_quality 为 not_scored。后两个字段刻意保留：引用 ID 存在不等于语义正确，离线替身也不能代表真实模型质量。

03 · 数据库设计：每个字段都对应一个故障问题

runtime.py · 实际使用的表结构

SCHEMA = '''
CREATE TABLE IF NOT EXISTS jobs(
 id TEXT PRIMARY KEY, spec TEXT NOT NULL, input_hash TEXT NOT NULL,
 status TEXT NOT NULL, generation INTEGER NOT NULL DEFAULT 0, owner TEXT,
 lease_until REAL NOT NULL DEFAULT 0, attempts INTEGER NOT NULL DEFAULT 0,
 available_at REAL NOT NULL DEFAULT 0, scope TEXT NOT NULL,
 memory_rev INTEGER NOT NULL, memory_expires REAL NOT NULL,
 deadline REAL NOT NULL, error TEXT);
CREATE TABLE IF NOT EXISTS checkpoints(
 job_id TEXT NOT NULL REFERENCES jobs(id), step TEXT NOT NULL, payload TEXT NOT NULL,
 PRIMARY KEY(job_id,step));
CREATE TABLE IF NOT EXISTS events(
 seq INTEGER PRIMARY KEY AUTOINCREMENT, job_id TEXT NOT NULL REFERENCES jobs(id),
 at REAL NOT NULL, generation INTEGER NOT NULL, kind TEXT NOT NULL, detail TEXT NOT NULL);
'''

下载完整 runtime.py

input_hash 由规范化 JSON 计算，包含材料、provider、model、endpoint、流程版本和作用域。重复提交时若摘要不同，直接报错。模型 Key 不进入 spec。恢复已有任务不会自动加载新版材料；需要明确创建新任务 ID。修改步骤语义时提升 report/v1 版本并使用新任务，这是本例的版本迁移规则。

若迁移到大规模服务，不应把大文档直接放进 jobs.spec。将对象存储 URI、内容哈希和授权快照存入任务；取回时核验哈希。此处只有两段合成材料，内联是为了降低复现实验的门槛。

04 · 认领与写回必须成对：只抢锁还不够

runtime.py · 原子认领

    def claim(self, job, owner):
        with transaction(self.db):
            row, now = self.job(job), self.clock()
            ready = (row['status'] in ('queued', 'retry_wait') and row['available_at'] <= now
                     or row['status'] == 'running' and row['lease_until'] <= now)
            if not ready:
                return None
            if row['deadline'] <= now or row['attempts'] >= 3:
                self.db.execute("UPDATE jobs SET status='failed',error='deadline_or_attempt_budget' WHERE id=?", (job,))
                self.event(job, row['generation'], 'failed', 'deadline_or_attempt_budget')
                return None
            generation = row['generation'] + 1
            self.db.execute("UPDATE jobs SET status='running',owner=?,generation=?,lease_until=?,"
                'attempts=attempts+1 WHERE id=?', (owner, generation, now + self.lease_seconds, job))
            self.event(job, generation, 'claimed', owner)
            return generation

下载完整 runtime.py

runtime.py · 写回前的执行者校验

    def guard(self, job, owner, generation):
        row = self.job(job)
        if row['status'] != 'running' or row['owner'] != owner or row['generation'] != generation or row['lease_until'] <= self.clock():
            raise LostLease('stale worker or cancelled task')
        return row

下载完整 runtime.py

SQLite 的 BEGIN IMMEDIATE 在事务开始时申请写事务；同一数据库仍只有一个写者，竞争时可能得到 SQLITE_BUSY。我们的认领事务只读改一条任务并追加事件，绝不在里面等待模型。连接设置 5 秒锁等待，超时后由调用方处理，不能把“用了事务”解释成永远不会争锁。

SQLite · Transaction：写事务的并发限制与 IMMEDIATE 行为来自官方文档。（核对：2026-09-25）

核心竞态可以手工推演：A 认领 generation=1；A 暂停到租约过期；B 认领 generation=2；A 恢复并提交。此时唯一检查点索引可能仍为空，所以单靠唯一键拦不住 A。guard 同时核验运行状态、owner、generation 和租约，A 的提交被拒绝。取消任务也递增 generation，让正在运行的旧工作者失去写回资格。

本例每步开始续租，不包含后台心跳。若一个模型调用超过 30 秒租约，即使结果成功返回也可能被拒收。这是明确的失败边界。接入生产时增加周期续租，并在续租失败后停止后续工具调用；外部工具还需自己的幂等或 fencing 协议。本地 guard 不能阻止已飞出的远端请求。

05 · 事务边界：工具在事务外，结果与事件在事务内

runtime.py · 提交结果与完成状态

    def commit_step(self, job, owner, generation, step, payload):
        with transaction(self.db):
            row = self.guard(job, owner, generation)
            self.validate_memory(row)
            if row['deadline'] <= self.clock():
                raise PermanentError('deadline exceeded')
            self.db.execute('INSERT INTO checkpoints VALUES (?,?,?)', (job, step, encode(payload)))
            self.event(job, generation, 'step_committed', step)
            if step == 'publish':
                self.db.execute("UPDATE jobs SET status='succeeded',owner=NULL,error=NULL WHERE id=?", (job,))

下载完整 runtime.py

一次 commit_step 同时保存步骤结果与 step_committed 事件。最后一步还把任务改为 succeeded，保证不会出现“任务成功但 publish 检查点没有落盘”的本地状态。恢复逻辑先读取已存在的检查点，只执行缺失步骤；它不要求模型根据一段日志猜测下一步。

LangGraph 的持久化文档描述了线程与检查点等机制。无论采用框架还是自行实现，都需要检查底层存储是否持久、哪些步骤会被回放、外部副作用如何去重。下面的实现属于本站设计，不是 LangGraph 内部源码。

LangGraph · Persistence：用于核对线程、检查点及持久化机制。（核对：2026-09-25）

06 · 故障实验 A：真正退出进程，再从 collect 之后恢复

复现命令

python3 cli.py submit --db crash.sqlite
python3 cli.py run --db crash.sqlite --lease-seconds 2 --fault after_collect
# 预期退出码 75；独立执行上面这条命令
python3 cli.py inspect --db crash.sqlite
# 等待至少 2 秒，让旧租约到期
python3 cli.py run --db crash.sqlite
python3 evaluate.py --db crash.sqlite

fault 分支在 collect 检查点提交后触发，CLI 调用 os._exit(75) 真正结束进程。它不会执行 finally 清理，也不会把任务改成 failed。因此首次 inspect 应看到 running 与 collect，而不是优雅异常处理后的理想状态。

如果第二次 run 仍输出 running，先看 lease_until：旧租约可能尚未到期，不要删除任务行强行恢复。若拖过默认 600 秒 deadline，任务会终止；使用新的 job ID 重新提交。本例认领次数上限为 3，首次执行也计数，不是“额外重试三次”。

07 · 故障实验 B：外部已成功，本地却没有回执

runtime.py · 独立服务的幂等协议

    def publish(self, job, payload):
        # Independent database simulates an external service's durable idempotency contract.
        db = sqlite3.connect(self.path + '.publisher.sqlite', isolation_level=None, timeout=5)
        try:
            db.execute('CREATE TABLE IF NOT EXISTS receipts(op TEXT PRIMARY KEY, payload_hash TEXT NOT NULL, receipt TEXT NOT NULL)')
            op, payload_hash = digest([job, 'publish/v1']), digest(payload)
            with transaction(db):
                old = db.execute('SELECT payload_hash,receipt FROM receipts WHERE op=?', (op,)).fetchone()
                if old:
                    if old[0] != payload_hash:
                        raise PermanentError('idempotency key reused with different payload')
                    return {'receipt': old[1], 'deduplicated': True}
                receipt = 'report-' + op[:12]
                db.execute('INSERT INTO receipts VALUES (?,?,?)', (op, payload_hash, receipt))
                return {'receipt': receipt, 'deduplicated': False}
        finally:
            db.close()

下载完整 runtime.py

操作键使用 SHA-256(job_id, publish/v1)，同一逻辑发布在重试时保持不变。服务保存 payload_hash 和 receipt；同键同内容返回旧回执，同键不同内容报错。不要把 generation 或重试次数放进操作键，否则每次接管都会绕过去重。

复现跨系统空隙

python3 cli.py submit --db effect.sqlite
python3 cli.py run --db effect.sqlite --lease-seconds 2 --fault after_effect
# 等待至少 2 秒
python3 cli.py run --db effect.sqlite
python3 cli.py inspect --db effect.sqlite

本次实际结果： 中断时有 collect / draft / verify 三个检查点；恢复后有四个。模拟发布库始终只有 1 条 receipts，恢复回执 deduplicated=true。查看原始运行快照。

这条测试验证的是：具有持久幂等契约的服务在本次故障下能返回同一个结果。它没有模拟网络分区、远端灾备、幂等键过期或服务实现错误。若接入的报告接口不支持幂等键，要先增加查询 operation_id 的能力；既不能查询也不能去重时，将超时记为 unknown，进入核对流程，不做盲目自动重试。

08 · 重试、取消与截止时间：把停止规则写成代码

这里的 2 秒、4 秒只是可复现实验参数，不是生产推荐值。实际退避可采用 full jitter：在 0 到 min(cap, base×2^attempt) 内随机取值；同时满足业务 deadline。仅增加重试次数可能提高表面成功率，也可能提高费用和重复业务影响。必须同时查看失败原因与最终业务记录。

API 适配器的 timeout=20 是 socket 层超时，不是严格的整个调用墙钟上限。慢速分块响应可能使调用时间更长；租约没有后台心跳时会进一步触发写回拒绝。若业务真的要求 60 秒内结束，需要可取消的 HTTP 客户端、整体 deadline 与任务调度协同。

09 · 接入千问或兼容接口：只替换 draft，保留执行边界

默认替身跑通后，才切换 provider=api。控制台中的模型 ID、地域、业务空间与 Key 必须匹配。适配器读取 LLM_API_KEY 环境变量，向你提供的 HTTPS 地址发送材料与有效记忆；它不把密钥写入任务数据库，也不跟随重定向。

先配置 LLM_API_KEY，再替换占位符执行

python3 cli.py submit --db api.sqlite --job api-001 \
  --provider api --model YOUR_MODEL_ID \
  --endpoint https://YOUR_TRUSTED_HOST/compatible-mode/v1/chat/completions
python3 cli.py run --db api.sqlite
python3 cli.py inspect --db api.sqlite

完整实现见 provider.py。请求固定 max_tokens=800、temperature=0，严格解析裸 JSON。若模型返回 Markdown 围栏，不做猜测性修复，而是显式失败，便于后续按错误样本调整提示词或采用服务支持的结构化输出。temperature=0 也不构成跨服务版本完全确定的承诺。

阿里云百炼 · Chat Completions：请求格式及地域地址以官方文档与控制台为准，文档列有业务空间专属域名。（核对：2026-09-25）

已用模拟 HTTP 成功响应和 429 响应验证适配器；未调用真实模型，也未测试供应商可用性。接入后仍需人工核对摘要是否受到原文支持。现在的 verify 只管结构与来源 ID。

10 · 放进已有 Java 系统：改哪些类、事务放在哪里

Java 接入顺序

// 结构示意；不是可直接编译的完整 Java 项目
var claim = claimService.claim(taskId); // 独立短事务
if (claim == null) return;
var input = checkpointService.loadInput(claim);
var output = modelClient.generate(input); // 不持有数据库事务
checkpointService.commit(claim, output); // 校验 generation + lease

不要在一个标了事务注解的大方法里完成“查任务 → 调模型 → 写结果”。长网络调用会长时间占用连接和锁。将短事务方法放入独立服务，通过真实事务边界调用；不要假定同对象内部方法调用必然触发框架代理。数据库引擎不同，认领 SQL 需要针对性实现和竞争测试。

若使用 PostgreSQL，可评估 FOR UPDATE SKIP LOCKED 的队列认领方式。它跳过锁定行，适合多个消费者争抢待处理任务；它不提供严格按创建时间完成的保证。你若要求某账户、某项目必须顺序处理，要另加分区串行约束，不能把 SKIP LOCKED 当 FIFO。

PostgreSQL · SELECT：官方说明 SKIP LOCKED 可用于多个消费者访问队列表，但不适合一般一致性视图。（核对：2026-09-25）

11 · 排障手册与上线前仍需补齐的部分

这份代码可以直接运行并复现故障，但没有 HTTP 服务、身份认证、生产队列、数据库迁移和监控部署；第三版补充了本地审批实验，尚非真实审批后台。生产前按具体业务补上这些部分，并测试磁盘满、远端成功后超时、网络分区、幂等记录保留期、数据库恢复以及权限变更。不要把基础 18 项通过写成生产可用率。

源码中的 scope 在 CLI 固定生成，默认信任本地操作者；绝不能直接映射为开放 API 请求体里的 tenant_id。真实写操作必须通过原业务权限服务，审批凭证应绑定动作类型、资源 ID、内容摘要、有效期与申请人。内容变化即重新审批，不能让模型拿旧批准执行新动作。

验收顺序建议为：先复现本页两个故障；再运行并发认领和旧代次测试；最后替换只读模型与报告沙箱接口。每一步保存任务库快照和远端回执。先能解释一次失败，再谈扩大并发。

12 · 审批要批准什么：不能只有一个 approved=true

继续用报告发布案例，把 publish 改成必须经过人工批准。最小实现很容易写成“用户点通过 → approved=true → 继续”。问题是人看过的草稿、实际发送的内容、所属项目或目标服务可能已经变化；布尔值没有记录批准范围，恢复进程也就无从核对。

approval_runtime.py · 真正参与审批摘要计算的字段

    def envelope(self, job, report):
        row = self.job(job)
        spec = json.loads(row['spec'])
        return {'job_id': job, 'action': 'report.publish/v1',
                'target': spec.get('report_target', 'local-report-store'),
                'scope': row['scope'], 'requester': json.loads(row['scope'])[1],
                'input_hash': row['input_hash'], 'payload_hash': digest(report),
                'memory_rev': row['memory_rev'], 'memory_expires': row['memory_expires']}

下载完整 approval_runtime.py

binding_hash 是上述动作描述的摘要，页面必须同时展示具体动作、目标和完整草稿。只显示一串 hash 让人批准，没有可理解的审查对象。Hash 也不是签名或登录凭证；操作者身份仍必须来自可信认证层。

OWASP · Transaction Authorization：授权应在服务端执行，重要交易数据应可供用户核对，并与执行阶段绑定。本文审批模型为独立教学实现。（核对：2026-09-25）

本例的 Principal 和 --actor 是本地测试夹具，不提供真实登录。开放成 HTTP 接口时，不允许浏览器提交 roles 或 allowed_scopes；这些字段应由认证会话和权限系统计算。local-report-store 是模拟目标标签，没有对接真实报告服务。

13 · 实验 D：保存草稿 → 等待审批 → 批准后继续

第三版使用独立入口 approval_cli.py，复用原 runtime.py 的采集、草稿、校验与检查点。基础 cli.py 仍保持原行为，方便比较是否增加审批。先下载首页第三版实验包，解压后运行：

建立待审批任务

python3 approval_cli.py submit
python3 approval_cli.py run
python3 approval_cli.py inspect

run 输出 waiting_approval。inspect 中 collect、draft、verify 已提交，publish 不存在。先读 checkpoints.draft 与 approval.action_json，再复制 approval.binding_hash，替换以下占位符。默认审批有效期 120 秒，任务总截止时间 600 秒；实验用短时间帮助观察过期，并非生产建议值。

确认同一份动作再恢复

python3 approval_cli.py approve --actor reviewer --hash REVIEWED_BINDING_HASH
python3 approval_cli.py run
python3 approval_cli.py inspect

等待期间没有工作线程一直 sleep，也没有数据库事务跨越人的审批时间。状态已经持久化，CLI 可以退出。审批决定用短事务写入，再把任务置为 queued；后续由本例第二次 run 或生产调度器接管。等待本身不占新的认领次数，首次执行与批准后恢复仍各算一次认领。

LangGraph 的 interrupt 也能持久化暂停并用同一线程恢复；但恢复时中断所在节点会从开头重跑，所以在 interrupt 之前的副作用仍需处理幂等。框架提供暂停机制，不等于替应用定义了审批范围或授权规则。

LangGraph · Interrupts：核对暂停、持久化、线程恢复以及节点重新执行的行为。（核对：2026-09-25）

14 · 审批回调要防什么：越权、重复、过期与内容变化

approve 和 reject 都需要同一个已审阅摘要。拒绝是终止决定，不能随后用旧请求再批准。需要修改内容时建立新任务，重新生成、审查；但如果旧任务已经发出，必须先核对它的结果，不能换新任务 ID 直接重复操作。

审批成功后、真正发出请求前，还要再核验一次。检查的不是“曾经批准过”，而是“此刻同一动作、同一内容、同一上下文仍然被允许”。验证和写入 dispatching 标记在同一个短事务中；下一节解释为什么这个标记需要独立状态。

approval_runtime.py · 发出前再次核验并持久化边界

    def publish(self, job, payload):
        if not self.execution or self.execution[0] != job:
            raise PermissionError('publish requires the guarded execution path')
        _, owner, generation = self.execution
        with transaction(self.db):
            row = self.guard(job, owner, generation)
            request = self.approval(job)
            if request is None or request['status'] != 'approved':
                raise PermanentError('approval required')
            self.validate_binding(job, request, payload)
            if request['expires_at'] <= self.clock() or row['deadline'] <= self.clock():
                raise PermanentError('approval_expired_before_dispatch')
            # Authorization cutoff: after this durable marker, outcome may be unknown.
            self.db.execute("UPDATE approvals SET status='dispatching' WHERE job_id=?", (job,))
            self.event(job, generation, 'dispatch_started', request['operation_key'])
        if self.fault == 'before_effect':
            raise InjectedCrash('dispatch marker committed; no service call was made')
        return super().publish(job, payload)

下载完整 approval_runtime.py

15 · 最难的窗口：请求已经发出，批准却过期了

考虑这个时序：审查通过 → 持久化 dispatching → 外部服务保存成功 → 本地进程退出 → 审批到期。恢复时若仅看到过期就标记“未执行”，会隐藏已经发生的业务结果；若重新申请批准再发送，也可能重复执行。授权是否允许新的写入，与过去的写入是否已经发生，必须分开回答。

实验 E：审批后的外部结果核对

python3 approval_cli.py submit --db approved-effect.sqlite
python3 approval_cli.py run --db approved-effect.sqlite
python3 approval_cli.py inspect --db approved-effect.sqlite
python3 approval_cli.py approve --db approved-effect.sqlite --hash REVIEWED_BINDING_HASH
python3 approval_cli.py run --db approved-effect.sqlite --lease-seconds 2 --fault after_effect
# 故障命令退出 75；等待至少 2 秒
python3 approval_cli.py reconcile --db approved-effect.sqlite
python3 approval_cli.py inspect --db approved-effect.sqlite

结果含义： effect_confirmed 表示确认了过去的写入；publish.fresh_write_performed=false 表示恢复只查询，没有发起新的发布。它不同于普通 succeeded，也不表示取消请求已回滚报告。查看本次原始记录。

将故障点换为 before_effect，会在 dispatching 提交之后、调用服务之前退出。测试注入器知道请求没发出，但恢复进程只看到 dispatching 和“无回执”，所以它停在 reconciling。这个保守选择牺牲自动继续的便利，换来不把“未知”错误当成“安全重发”。真实系统若有可靠的服务端幂等重试契约，可以另行设计恢复协议；本扩展只实现只读核对。

没有回执的情况要配置人工核对或远端终态查询，不能无限停着无人处理。当前代码没有后台核对器、告警或人工结案接口。生产中结案也必须保留证据，禁止运维直接把 dispatching 改回 queued。

16 · 取消的分界线：停止后续动作，不伪造回滚结果

派发前的最后授权检查是本例的分界线。取消与 dispatching 更新竞争同一数据库短事务：取消先提交则工作者失去资格；dispatching 先提交则进入结果未知区域。外部请求不处在本地事务中，因此不能提供“点取消后绝无业务影响”的全局保证。

补偿不是恢复代码的通用反向函数。删除草稿、撤销发布、退款、冲正，分别有不同业务规则。接入 Java 业务服务时应由原服务提供明确补偿命令、独立审批和审计，不让模型根据自然语言随意构造反向 SQL。

17 · Java 接入：把审批、派发和核对拆成明确服务边界

推荐 API 合同是 POST /approvals/{id}/decision，请求体只包含 decision、reviewed_hash 和客户端请求幂等 ID。actor 从会话读取，权限与租户范围从服务端计算。返回 409 表示版本或决定冲突，403 表示无权审批，410 可用于已过期请求；具体状态码是本页建议，当前实验没有 HTTP API。

本次扩展仍是本地审批状态机与身份夹具，没有实现登录后台。迁移时先用现有登录和角色服务接入 read-only 业务沙箱，再验证：过期批准不可发送、内容变更重新审查、取消后业务效果如实显示、回执不明不创建第二份操作。

来源与验证

官方资料用于核对具体机制；表结构、程序与实验为本站独立设计。

验证记录 · 审批恢复证据