项目：可恢复任务工作流

项目目标

用 Python 标准库实现并验证一个离线工作流：规则路由、只读检查并行扇出与汇聚、高风险审批、拒绝重复键的严格 JSON 检查点、错误分类、审批上下文指纹、幂等回执和“动作提交后崩溃”恢复。

脚本是 00-初学者路线/examples/resilient_workflow.py，测试是 00-初学者路线/examples/test_resilient_workflow.py。它不调用模型、网络或真实业务系统，不需要密钥。项目演示的是架构语义，不是生产审批、分布式锁或持久化方案。

你将看到的状态机

start -> checks -> evaluate
  ├─ 检查未通过 -> failed
  ├─ low        -> execute -> done
  └─ high       -> awaiting_approval
                    ├─ reject  -> canceled
                    └─ approve -> execute -> done
 
checks 失败 -> failed
execute 提交后崩溃 -> 仍停在 execute -> 根据 receipt 恢复 -> done

两个检查 worker 不修改共享状态；主线程 join 后一次写入结果。高风险路径先完成两个只读检查，再暂停展示审批；审批指纹绑定规范化动作、通过的检查证据、风险和 policy_revision，因此这些输入改变会拒绝旧批准。动作通过稳定 action_id 写本地回执，恢复时若回执一致就复用，不重复提交。

教学边界

检查点哈希不含密钥，只能检测意外损坏或普通篡改，不提供身份认证。解析器拒绝重复 JSON 键，schema 版本与 policy_revision 不兼容时失败关闭；这仍不等于数据库事务、签名或迁移方案。生产系统仍需并发控制、访问控制、加密、审计保留、显式迁移和灾备。

环境准备

在 Agentic Design Patterns 目录打开 PowerShell 7。使用当前稳定 Python 3；本项目无第三方依赖，不需要虚拟环境。创建一次性临时目录：

$demo = Join-Path $env:TEMP ("agentic-pattern-" + [guid]::NewGuid())  # 为本轮实验生成唯一的临时产物目录。
New-Item -ItemType Directory -Path $demo | Out-Null  # 创建目录并隐藏无关的 PowerShell 对象输出。
$script = Resolve-Path .\00-初学者路线\examples\resilient_workflow.py  # 解析工作流脚本的绝对路径，避免受当前目录影响。

$demo 位于当前用户临时目录，不会向 vault 写运行产物。

实验一：低风险直接完成

# 使用低风险输入运行完整工作流；反引号只把同一条命令拆成多行展示。
python -B $script `
  --checkpoint "$demo\low-state.json" `
  --receipt "$demo\low-receipt.json" `
  --task-id "low-001" `
  --risk low  # 低风险路径无需人工批准，应产生状态与回执。

预期终态为 done，事件包括路由、并行汇聚、验收通过与动作提交。打开两个 JSON 文件，检查状态与动作回执是否使用相同 task_id 和动作指纹。

实验二：高风险暂停、批准与拒绝

第一次运行不提供决定：

# 首次高风险运行故意不提供决定，以验证流程会安全地暂停等待审批。
python -B $script `
  --checkpoint "$demo\high-state.json" `
  --receipt "$demo\high-receipt.json" `
  --task-id "high-001" `
  --risk high  # 声明高风险动作，触发 awaiting_approval 分支。
$LASTEXITCODE  # 输出预期的等待审批退出码，确认没有误执行写动作。

预期终态 awaiting_approval，退出码为 3，且回执文件不存在；此时 checkpoint 已含完整的只读检查结果。脚本拒绝在首次调用直接附带 --decision approve，以免把命令行参数误当成已展示过的人工批准。使用同一 task、risk 和检查点恢复：

# 对同一 checkpoint 提供明确批准，验证恢复时重新核对输入与审批指纹。
python -B $script `
  --checkpoint "$demo\high-state.json" `
  --receipt "$demo\high-receipt.json" `
  --task-id "high-001" `
  --risk high `
  --decision approve  # 以显式人工决定恢复暂停的高风险任务。

预期终态 done。另起一组新路径，把 approve 改为 reject，预期终态 canceled、退出码 4、无回执。拒绝不会被当成可重试错误。

实验三：分类重试

模拟 policy 检查第一次出现瞬时超时：

# 注入一次可分类的瞬时故障，检查有限重试而非无限循环。
python -B $script `
  --checkpoint "$demo\retry-state.json" `
  --receipt "$demo\retry-receipt.json" `
  --task-id "retry-001" `
  --risk low `
  --simulate-transient-once  # 仅第一次 policy 检查返回瞬时错误。

预期完成，attempts.policy 为 2。再用新路径加 --simulate-permanent-policy-failure，预期 failed、退出码 1，policy 只执行一次且不创建回执。这证明“是否重试”取决于类别，不取决于失败文本长短。

实验四：动作已提交但检查点未更新

# 在动作已提交但 checkpoint 尚未更新时注入崩溃，模拟最危险的恢复窗口。
python -B $script `
  --checkpoint "$demo\crash-state.json" `
  --receipt "$demo\crash-receipt.json" `
  --task-id "crash-001" `
  --risk low `
  --crash-after-commit  # 让脚本在写入回执后退出，保留恢复所需证据。
$LASTEXITCODE  # 输出教学用崩溃退出码，而不是把它误认作正常完成。

预期退出码 5。此时 receipt 已存在，而 checkpoint 仍停在 execute。去掉模拟开关，使用完全相同参数再次运行：

# 以完全相同的业务参数重跑，验证工作流复用已有回执而不是二次提交。
python -B $script `
  --checkpoint "$demo\crash-state.json" `
  --receipt "$demo\crash-receipt.json" `
  --task-id "crash-001" `
  --risk low  # 不再注入崩溃，使恢复逻辑有机会完成终态写入。

预期事件为 action:recovered_existing_receipt，不是第二次提交。若把另一个任务的 receipt 放到该路径，程序会拒绝覆盖。

退出码契约

退出码	含义
0	done
1	确定性检查失败
2	输入、检查点、回执或未持久化的审批决定不安全
3	等待审批，稍后可恢复
4	人工拒绝，任务已取消
5	教学用“提交后崩溃”

自动化调用方应根据退出码和结构化 JSON 共同判断，不解析自然语言日志。

阅读代码的顺序

1. new_state / validate_state：严格 schema、版本、策略版本和不变量。
2. save_state / load_state：封装哈希、拒绝重复 JSON 键与原子替换。
3. route：规则优先的风险路由。
4. run_parallel_checks：worker 无共享写入，主线程统一 join。
5. approval_fingerprint_for：把高风险批准绑定动作、证据、风险与策略版本。
6. perform_action：稳定 action ID 与回执恢复。
7. run：先检查、再审批、再执行的显式状态转换。
8. main：CLI 参数、JSON 输出和退出码。

运行测试

在 examples 目录执行：

Push-Location .\00-初学者路线\examples  # 进入示例目录，使测试可直接导入同目录模块。
$env:PYTHONDONTWRITEBYTECODE = "1"  # 禁止测试产生 .pyc 缓存文件。
python -B -m unittest -v test_resilient_workflow.py  # 在普通模式运行完整工作流回归测试。
python -B -O -m unittest -v test_resilient_workflow.py  # 验证关键校验不依赖优化模式会移除的裸 assert。
python -B -W error -m unittest test_resilient_workflow.py  # 将 warning 视为失败，暴露潜在兼容问题。
Pop-Location  # 恢复进入 examples 目录前的工作位置。

测试覆盖严格状态校验、重复 JSON 键、策略版本、路由、并行汇聚、瞬时与永久错误、检查通过后才暂停、审批上下文变化、首次调用审批绕过、批准、拒绝、完成后重复恢复、提交后崩溃和 CLI 退出码。unittest 断言在 -O 下仍有效。

综合任务

在不降低现有测试的前提下，增加第三个只读分支 compliance：

1. 更新分支常量、默认检查和严格 schema；
2. 让汇聚器拒绝缺少该分支；
3. 增加 compliance 瞬时失败、永久拒绝和完整成功测试；
4. 解释它为何可与另外两个分支并行；
5. 记录测试数量与正常、-O、warnings-as-errors 三种运行结果。

完成标准不是“代码能启动”，而是新分支缺失时必须失败关闭、已有恢复语义不变。

项目验收

• 低风险路径达到 done，回执与任务匹配。
• 高风险无决定时暂停，拒绝不产生回执。
• 高风险策略检查失败时不会进入审批；审批绑定当前动作、证据、风险和策略版本，任一变化会被拒绝。
• 首次调用不能把 --decision 当成已经展示过的人工批准。
• 只有 transient 检查在预算内重试。
• 检查点损坏、重复键、未知字段和版本/策略不符都失败关闭。
• 提交后崩溃能凭相同回执恢复，不重复动作。
• 正常、-O 与 warnings-as-errors 测试均通过。
• vault 中没有检查点、回执、缓存、密钥或真实凭据。

与课程的对应

• 路由与汇聚：路由、并行与汇聚
• 审批与指纹：人工审批与安全边界
• 错误、回执与评测：失败恢复、评测与可观测性

参考资料

• LangGraph 官方：Interrupts（恢复与幂等注意事项；核对：2026-07-14）
• LangGraph 官方：Persistence（检查点概念；核对：2026-07-14）
• Anthropic：Demystifying evals for AI agents（评测对象；核对：2026-07-14）
• MCP 2025-11-25：Authorization 与 A2A 1.0：Security Considerations（真实协议边界仍须由生产实现做身份、对象范围与 token 验证；核对：2026-07-22）