本篇直接读取上一篇的runs/demo-001/receipts/*.json并复用首篇延续至今的events.jsonl、086 的预算配置和状态推导函数。replay.py把工具调用替换成回执查询输出replay_report.json。上一篇回执是本篇工具结果源因此重放不会再次修改文件或访问外部系统。一、重放的目的不是复刻模型随机性同一个提示再次请求模型采样、模型版本和服务端实现都可能变化这叫重新运行不叫重放。工程重放关注宿主是否在相同已记录输入上做出相同授权、状态转换和停止判定。模型建议作为事件保存工具结果由回执提供确定性控制面可以完全离线验证。读取器首先检查事件序号、任务编号、已引用回执是否存在再按事件类型归约状态。未知事件不能悄悄跳过新版本读取旧事件可以声明兼容旧读取器遇到新语义则应失败避免生成貌似成功的残缺报告。from__future__importannotationsimporthashlibimportjsonfrompathlibimportPathfromtypingimportAnyfromledgerimportload_eventsfrommachineimportderive KNOWN{run_started,model_proposed,tool_allowed,tool_denied,attempt_started,tool_started,tool_finished,model_finished,check_finished,state_changed,}defload_receipts(root:Path)-dict[str,dict[str,Any]]:receipts{}forpathinsorted(root.glob(*.json)):datajson.loads(path.read_text(encodingutf-8))ifdata[call_id]!path.stem:raiseValueError(fbad receipt name:{path.name})receipts[data[call_id]]datareturnreceiptsdefreplay(root:Path)-dict[str,Any]:eventsload_events(root/events.jsonl)unknownsorted({event[kind]foreventinevents}-KNOWN)ifunknown:raiseValueError(funknown events:{unknown})run_ids{event[run_id]foreventinevents}iflen(run_ids)!1:raiseValueError(mixed run ids)receiptsload_receipts(root/receipts)foreventinevents:call_idevent[payload].get(call_id)ifevent[kind]tool_finishedandcall_idnotinreceipts:raiseValueError(fmissing receipt:{call_id})budgetjson.loads(Path(budget.json).read_text(encodingutf-8))statederive(events,budget)source_hashhashlib.sha256((root/events.jsonl).read_bytes()).hexdigest()return{run_id:next(iter(run_ids)),events:len(events),receipts:len(receipts),state:state.name,events_sha256:source_hash}defmain()-int:rootPath(runs/demo-001)reportreplay(root)(root/replay_report.json).write_text(json.dumps(report,indent2),encodingutf-8)print(fevents{report[events]}receipts{report[receipts]}state{report[state]})return0if__name____main__:raiseSystemExit(main())运行输出events14 receipts1 stateretrying二、为什么重放要严格而不是尽量恢复面向用户数据的导入器可以跳过一条坏记录并报告异常审计重放却不能如此。缺一条tool_finished可能改变费用与状态忽略后得到的结论没有证明力。严格模式在第一处不一致停止并指出序号、事件类型和工件。恢复工具可以另做“尽力读取”用于提取损坏前缀但输出必须标记partial不得与完整重放报告混用。记忆点是恢复回答还能拿回什么重放回答当时为何这样决定。两个问题的可信度要求不同。importjsonfrompathlibimportPathdefcompare_reports(left:Path,right:Path)-list[str]:ajson.loads(left.read_text(encodingutf-8))bjson.loads(right.read_text(encodingutf-8))keys(run_id,events,receipts,state,events_sha256)return[keyforkeyinkeysifa.get(key)!b.get(key)]rootPath(runs/demo-001)firstroot/replay_report.jsonsecondroot/replay_report.second.jsonsecond.write_text(first.read_text(encodingutf-8),encodingutf-8)changedcompare_reports(first,second)print(fdeterministic{notchanged}changed_fields{changed})运行输出deterministicTrue changed_fields[]三、故障注入让恢复路径真的被走到正常测试很难覆盖进程恰好在fsync后退出。可以在关键边界增加仅测试可用的故障点事件已写、回执未写临时文件已写、原子替换未做检查完成、状态未缓存。测试进程收到预定异常后重新启动再断言事件推导结果。故障点名称要稳定并写入测试配置不能在生产环境接受模型传入。随机杀进程适合发现未知窗口确定性故障点适合回归。两者结合先用随机试验发现问题再把具体窗口固化成可重复测试。四、重放报告如何用于排障报告包含输入账本摘要、契约摘要、回执数量、最终状态和读取器版本。它不包含全部源码却能确认客服收到的事件文件是否与客户现场一致。如果同一摘要在两台机器导出不同状态说明状态机实现或运行版本有差异。本篇产物replay_report.json是下一篇对抗测试的基线。下一篇会系统篡改测试、目标、工具清单、事件与回执验证 ProofLoop 不会把这些攻击误判为成功每个变异都必须被某个确定性守卫捕获。五、并发事件怎样获得可重放顺序单进程用自增序号足够多进程同时追加则可能争用同一个next_seq。不能让每个进程先数行再写因为两者会生成相同序号。可由单独账本进程分配序号或使用数据库事务中的自增主键。工具内部并发产生的子事件还可带parent_seq把全局提交顺序与局部因果关系分开。全局顺序不一定代表现实动作的精确先后。例如两个网络请求并行完成谁先写账本只说明宿主先观察到谁。回放控制面时按提交序号处理分析业务因果时使用调用 ID、父事件和回执。把这两种顺序混为一谈会从日志推导出并不存在的依赖。六、版本升级不能偷偷改变历史结论状态机新版本可能修复旧算法。重放报告必须记录读取器版本与规则摘要同一版本对同一输入必须一致新版本若给出不同状态应并排生成迁移报告解释哪条规则改变。不能覆盖旧报告后声称任务当时就是新结论。事件 Schema 演进采用“新增可选字段优先”改变字段含义则提升版本。迁移器输出新事件流和旧流摘要不原地编辑证据。记忆点是重放可以有新解释但历史必须保留旧字节。这使审计既能使用修复后的规则也能还原当时系统为何做出原决定。对包含敏感输出的回放验证器应只读取摘要和必要字段不能为了方便把所有工件上传到集中服务。离线重放的价值之一就是把排障能力带到数据所在位置只带走不含正文的报告。参考来源Martin FowlerEvent SourcingPython 文档unittest — Unit testing framework 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 本文属于《Agent可靠性工程实战》系列持续更新关注不迷路。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。