后台任务、检查点与发布回归
长任务不能依赖一条 HTTP 连接存活。入口持久化目标后返回 ID；队列投递 ID；worker 领取租约、执行一小步、提交结果。队列通常允许重复投递，因此 checkpoint 记录的不只是模型消息，还包括业务状态、输入版本、已发生副作用和恢复位置。
过期 worker 为什么不能提交
完整标准库实验用单调递增的 fencing token 表达每次领取权。它是并发控制的最小模型；数据库实现需要把比较与更新放在一个原子操作里，不能用下面的进程内字典代替分布式锁。
state = {"epoch":0, "status":"queued", "result":None}
def claim():
state["epoch"] += 1
state["status"] = "running"
return state["epoch"]
def commit(epoch, value):
if epoch != state["epoch"] or state["status"] != "running":
return False
state.update(status="completed", result=value)
return True
old = claim()
new = claim() # 模拟旧租约过期后新 worker 接管
assert not commit(old, "stale")
assert commit(new, "fresh")
assert state["result"] == "fresh"
assert not commit(new, "duplicate")
print(state)旧 worker 即使最后才返回，也不再拥有提交权。超时停止等待不能证明旧任务已停止，fencing 保护结果提交；外部副作用仍必须由上一章的幂等账本保护。取消可把状态改为 cancelled，使之后提交失败，但不能撤销此前已经发送的邮件。
恢复时重新检查什么
检查点应保存 schema_version、图版本、租户、任务 ID、状态、证据引用、审批摘要及效果状态。恢复前检查权限是否撤销、文档是否变更、工具是否还存在。旧版本 checkpoint 不能直接按新字段解释；先写纯函数迁移，保留原副本，用旧输入和新输入共同回归。
RAG 作为上下文注入时，每轮固定取得证据；作为工具时，模型决定查询时机。精确库存走授权业务 API，语义资料走检索；两条路径都返回来源与版本。权限必须在召回前限制候选，并在返回前复验，缓存键纳入租户、权限版本、索引版本与查询。
从恢复测试到部署
离线 golden 集包含成功、空资料、工具超时、撤权、重复领取、取消与注入。任务成功率按业务断言统计；另测工具选择、引用、重复次数、token 与 p95 延迟。灰度先限定租户与风险，比较同类请求切片，错误或成本超阈值熔断为只读/人工路径。健康检查确认数据库与队列，不能只看进程存活。
事件驱动、层级 Agent、代码 Agent 与多模态输入都复用上述任务协议：事件至少有 task ID、序号、版本与幂等键；子 Agent 不继承超出任务的权限；代码执行结果须包含差异与测试证据；图像和音频需保存来源及处理版本。框架升级与模型升级分开灰度，才能判断回归来自哪一层。
任务账本怎样跨进程接续
内存token只解释提交权限。实际后台任务需要先写入任务身份和状态，再回复调用方任务ID；worker持有租约代数，阶段产物落库后才能推进step。下面SQLite账本通过两个独立子进程演示：第一次只完成检索并退出，第二次从step=1继续生成；同一任务不会重新从0开始。
保存为jobs.py运行。例中claim代表已通过调度器判定旧租约过期后的重新领取；为聚焦fencing，未实现时钟租约或生产并发调度，不能让活跃worker任意互抢。
import sqlite3, sys, subprocess
from pathlib import Path
from tempfile import TemporaryDirectory
def database(path):
db=sqlite3.connect(path)
db.execute('CREATE TABLE IF NOT EXISTS jobs(id TEXT PRIMARY KEY, step INTEGER, epoch INTEGER, status TEXT, result TEXT)')
return db
def claim(db, job):
# 单事务发布新租约代数；生产还需lease_until，过期后才允许重新领取。
with db:
db.execute("UPDATE jobs SET epoch=epoch+1,status='running' WHERE id=? AND status IN ('queued','running')",(job,))
row=db.execute('SELECT step,epoch,status FROM jobs WHERE id=?',(job,)).fetchone()
if row[2]!='running':raise ValueError('terminal task')
return row[:2]
def save(db, job, epoch, step, result, terminal=False):
with db:
cur=db.execute("UPDATE jobs SET step=?,result=?,status=? WHERE id=? AND epoch=? AND status='running'",(step,result,'completed' if terminal else 'running',job,epoch))
return cur.rowcount==1
def worker(path):
db=database(path);step,epoch=claim(db,'j1')
if step==0:
assert save(db,'j1',epoch,1,'retrieved evidence')
else:
assert save(db,'j1',epoch,2,'answer from persisted evidence',True)
db.close()
if __name__=='__main__':
if len(sys.argv)==2:
worker(sys.argv[1])
else:
with TemporaryDirectory() as folder:
path=str(Path(folder)/'jobs.sqlite');db=database(path)
with db:db.execute("INSERT INTO jobs VALUES ('j1',0,0,'queued','')")
for _ in range(2):subprocess.run([sys.executable,__file__,path],check=True)
assert db.execute("SELECT step,status FROM jobs WHERE id='j1'").fetchone()==(2,'completed')
with db:db.execute("INSERT INTO jobs VALUES ('cancel',0,0,'queued','')")
_,old=claim(db,'cancel');_,new=claim(db,'cancel')
assert not save(db,'cancel',old,1,'stale')
with db:db.execute("UPDATE jobs SET status='cancelled',epoch=epoch+1 WHERE id='cancel'")
assert not save(db,'cancel',new,1,'late')
db.close()
print('resume, fencing, cancellation passed')条件UPDATE把epoch与running同时写进提交前提，陈旧worker更新0行；取消既改终态又提升epoch，旧worker即便仍运行也不能发布。进程崩溃可能发生在外部写操作已完成、save尚未执行之间，因此有副作用步骤必须复用前章幂等键，SQLite事务本身无法对远程系统实现原子提交。
查询接口从账本返回status/step，绝不因为worker退出就称任务完成。读取已完成任务应直接复用结果；对于已取消/完成，claim拒绝。真实服务还要限制任务归属、调度重试次数与保留期。
练习
账本变体：首次worker结束后取消j1，再启动恢复worker，必须拒绝领取；确认持久step仍为1，不能发布最终结果。
在 claim 后取消任务，再让当前 epoch 提交，必须失败且 result 保持 None；随后重新排队并领取新 epoch 可以成功。