多 Agent 消息协作与重启恢复
消息收到之后，怎样让工作在崩溃、重连和重复投递下继续保持可解释的状态。用一个完全离线的实验，逐项验证任务与回复的恢复规则。
你能带走什么
- 收件 ACK 与业务终态分开记录
- 结果和 outbox 在本地事务中一起提交
- 租约回收失联任务，fencing 拒绝旧提交
- 至少一次思路配合明确范围的幂等
- 宿主触发能力与人类授权分别核验
1. 先把可恢复的协作定义清楚
把实现、审查和浏览器验证交给不同 Agent，可以减少一个会话同时处理多种工作的负担。真正进入持续协作之后，任务还会遇到进程退出、重复消息、回执丢失、两个执行器同时抢活，以及人类已经取消但旧任务继续提交等问题。只保存聊天记录，无法回答这些问题。
本篇沿用一个具体分工：协调器派发检查任务，执行器产生结构化结果，审查者核对证据，发布动作由单独的授权入口完成。共享仓库保存代码与产物；持久数据库保存谁接了什么、当前租约、尝试次数和待发送回复。不同执行器使用隔离目录或 worktree，合并与发布保留单一负责人。
下面的 Python 示例只用标准库和 SQLite。任务处理器返回确定性结果；两个数据库文件分别模拟本地队列和远端接收服务。没有真实模型调用、HTTP 请求、通知、生产任务或外部 Agent 通信。所有故障均由测试显式注入，读者可以先验证恢复规则，再考虑接入自己的宿主。
目标是让重复到达的同一请求收敛到可检查的状态。在有限重试预算内，投递采用 at least once 的处理思路，允许重复尝试；幂等效果只覆盖示例明确维护的去重记录和同事务业务记录。持续故障、记录过期、外部工具不支持去重时，都不能承诺无条件恰好一次。
图 01 / 离线架构示意
查看图示文字说明
协调器经认证入口将消息写入 inbox 和 tasks。worker 认领任务，提交结果及 outbox。发件器向独立 receipts 数据库投递。模型、网络和真实对外工具均未接入。
2. 先固定任务身份和完成契约
一个可重试请求需要稳定身份。示例以认证主体与 message_id 的组合区分入站消息，再派生内部 task_id。message_id 在重发、重连补收和进程重启后保持不变；发送重试时临时生成新 ID，会把同一工作变成多个合法任务。不同主体允许使用相同的本地消息编号，避免租户之间互相去重。
Envelope 还保存 payload、deadline、max_attempts、hops 和 kind。相同身份的重复请求必须带回相同语义内容，因此示例保存规范化 JSON 的内容指纹；同一 ID 对应不同内容时直接报冲突。业务参数、截止时间或权限范围发生变化，应创建明确的新请求，并保留它与原请求的关联。
实际项目可以再加 correlation_id、parent_task_id、输入版本、工件地址与结果模式。验收条件应具体到文件哈希、测试命令和预期状态。聊天中一句已经完成，不能替代这些证据。结果包含问题列表时，也要区分检查流程成功和被检查对象通过，两者可能得出不同结论。
幂等键的保存期也属于契约。Stripe 的官方幂等文档明确说明键可能在保留期后清理，旧键再次使用会产生新请求，且参数冲突会被拒绝。这里引用的是边界设计经验，示例不调用 Stripe；生产系统应公布自己的保留期，并让它覆盖可能发生的最长重放窗口。
3. 收件确认和业务完成分两次回答
接收函数先校验来源与结构，再在同一事务内插入 inbox 和 tasks，提交成功后返回收件 ACK。此时系统承担了保存并处理这条任务的责任，但 worker 可能尚未启动。若在提交之后、返回 ACK 之前退出，发送端会重试；接收端查到已有记录，就返回同一个任务身份。
完成结果来自另一个阶段。worker 获得有效租约后运行，校验输出，再把任务终态与 outbox 记录一起提交。发送器随后交付该结果。监控界面至少分别显示收到、处理中、业务终态和结果投递状态，不能用一个绿色勾把四个阶段合并。
当前 OpenAI MCP Events 官方接入文档同样明确：2xx 确认 webhook 收件，ChatGPT 异步处理事件；事件可能乱序，写入工具应具有幂等性。本篇从完成收件之后展开，不重复签名与回调验证实现。具体宿主允许怎样唤起一次执行，需要单独确认。
图 02 / 离线架构示意
查看图示文字说明
发送端传入固定消息 ID；接收端将 inbox 和 task 提交后返回收件确认；worker 此后才运行并提交结果与 outbox；最终结果是第二类消息。不能从收件确认推断业务成功。
4. 把任务和发件箱各自建成状态机
示例的 inbox 保存已经接受的原始任务和指纹；tasks 保存业务执行状态、owner、lease_until、fence、attempts 和结果；outbox 保存稳定回复 ID、回复内容及投递进度。远端模拟器则有独立 receipts 和效果记录。表的约束负责拒绝重复身份，事务负责保证相关写入一起成功或一起撤销。
任务从 pending 进入 running，随后进入 done、failed、cancelled 或 expired；输入达到循环跳数边界时进入 hop_limit，保留终态证据。可恢复故障会让任务回到 pending，但次数和截止时间继续累计。发件箱有自己的 pending、sending、sent、dead 状态。业务成功以后，投递仍可能失败，不能因此再次运行已完成的业务。
使用 SQLite BEGIN IMMEDIATE可以在认领前进入写事务。同一数据库同时只有一个写事务，其他连接可能等待或收到 SQLITE_BUSY。因此认领事务只读写少量状态，模型调用、浏览器操作和网络发送必须放在事务之外；持锁等待数分钟会把整个队列堵住。
处理器还需要区别可重复计算与不可逆动作。计算报告时，可以让两个尝试重复读取同一输入版本，再由有效租约提交一份结果；发送通知、创建工单或修改账号时，应先设计目标系统的幂等身份。把后者藏在一个普通 Python 函数里，不会自动获得事务保护。本文故意把这些真实副作用排除，便于读者检查恢复不变量。
示例启用 WAL、外键与同步写入配置，并使用磁盘临时目录做恢复测试。这能验证程序约定的事务边界，不能证明所有磁盘、文件系统和断电场景都无损。真实部署还要设计备份、容量、水位清理及恢复演练；本示例的 WAL 数据库要求进程位于同一主机，不适用于跨主机共享网络文件系统。
图 03 / 离线架构示意
查看图示文字说明
任务由pending进入running；成功提交进入done；可重试失败回pending，耗尽进入failed；取消与超时阻止后续提交。入队之前的拒绝独立于任务状态。
5. 把回复和结果写进同一个事务
最常见的双写顺序是先改业务状态，再发送消息。若进程刚完成第一步就退出，任务显示成功，但接收方永远得不到结果。反过来先发送再记成功，也会留下消息已到达、本地却准备重新执行的窗口。outbox 将业务结果和待发送意图放入一个本地事务，随后由独立发送器读取并投递。
本文为每个任务的终态回复派生固定 reply_id。发件器无论重试多少次，都发送这条已保存的回复内容。远端以 reply_id 做唯一键，同时比较内容指纹；首次接收把回执与模拟业务效果一并提交，重复接收只返回已有结果，内容变化则拒绝。AWS 的 transactional outbox 指南也提醒消费端处理重复消息。
跨系统仍有一个不能消失的窗口：远端已经提交，ACK 在返回途中丢失，本地看起来像发送失败。正确恢复动作是复用 reply_id 重发，并让远端去重。若真实目标是一个没有幂等接口的邮件服务，重复投递可能生成两封邮件；在本地表里加 UNIQUE，无法单独消除这类外部副作用。
发送器自己也需要 claim、租约和 fencing。否则两个进程可能同时读取 pending 回复并反复投递。到达重试上限后进入 dead，保留最后错误和原始身份，交由可审计的人工处理。dead 表示发送端在预算内未取得可信确认，远端有可能已经完成接收；人工重放前应先按 reply_id 对账。重置次数、删掉去重记录或换一个 ID 强行重发，都应被视为新决策。
图 04 / 离线架构示意
查看图示文字说明
四个故障点分别位于本地提交前、本地提交后、远端提交后但本地确认前、本地记录发送完成后。每个点的恢复动作不同。远端提交后重发必须复用稳定回复 ID。
6. 用租约回收失联执行器，用 fencing 拒绝旧提交
单纯把状态改成 running 会让崩溃任务永久卡住；每次启动就把全部 running 改回 pending，又可能抢走仍在工作的任务。租约给所有权一个明确期限：claim 在短事务内选中可执行记录，写入 owner 和 lease_until，增加 attempts，同时增加单调 fencing token。
假设 A 获得 token 1 后失联，租约到期，B 重新认领并获得 token 2。A 恢复时仍可能带回一个看似正确的结果。提交函数必须核对 owner、token、租约和当前状态，拒绝 A 的旧写入。取消、截止时间和结果提交的竞争，也通过同一数据库的短事务按实际提交顺序裁决。
长任务可在租约到期前续租。续租只能由当前有效拥有者执行，不能让旧 token 复活。租约长度应覆盖正常心跳间隔和短暂延迟，但不宜长到故障后等待数小时。Amazon SQS visibility timeout 文档提供了类似的可见性窗口模型，也明确重复投递仍可能发生。
fencing 的效果取决于谁检查它。本文由 SQLite 提交函数检查，因此保护的是本地任务和 outbox；若 A 在失联前已调用另一个系统修改数据，B 的新租约不会撤回该修改。外部写入要有稳定业务幂等键，或让目标服务验证单调版本。无法取得这类保证时，恢复前应先查询目标状态，必要时停下来人工对账。
图 05 / 离线架构示意
查看图示文字说明
A持有token1后失联，租约到期后B用token2重新认领；A恢复时尝试提交token1被数据库拒绝。该机制保护本地状态，不能自动撤回外部操作。
7. 重启时恢复状态，重连时逐项补收
重启流程先打开原数据库；生产系统还应校验 schema 版本和迁移兼容性。接着整理已经到期的租约与截止任务，随后分别启动 task worker 和 outbox dispatcher。不要清空表，也不要仅凭最后一条聊天内容重新生成全部工作。已经提交的结果继续用于发送；尚未提交的任务等待租约到期后重领。会话摘要和原始输入可以帮助模型理解上下文，但不能改变数据库已经确认的终态。
批量补收需要规定部分成功语义。示例 receive_batch 逐条提交，第三条失败时，前两条仍然存在，并通过异常携带已提交的 ACK 前缀。调用方保存这个前缀，修复失败项后再继续。若选择整批一个事务，则要明确整批回滚；不能把这两种接口的返回约定混在一起。
上游有 cursor 时，只推进到已持久确认的连续位置。不能因为第十条成功，就跳过尚未处理的第九条。消息乱序时可单独保存已接收集合和连续水位。本文的批次测试模拟重连后的重复到达，不实现某个供应商的 cursor 协议，也不假装能补回服务端已经清理的历史。
对每一种实际传输都要确认回放能力、保留窗口、顺序和缺口信号。MCP Events 草案和具体宿主支持范围可能不同；当前 OpenAI 接入页面声明不支持其 polling、streaming、gap 与 terminated 通知。遇到缺口时应报告数据不完整，再用获准的查询接口做快照校对。
图 06 / 离线架构示意
查看图示文字说明
四条消息批次中，101已提交，102冲突失败，103和104尚未处理。调用方只可记录成功前缀，修复冲突后恢复；重投101通过去重返回。
8. 把超时 取消 重试与循环预算一起保存
每个任务至少要有三个时间边界：单次调用超时、认领租约和绝对 deadline。它们分别限制一次等待、拥有权和整个任务寿命。重试不能重新生成 deadline，否则长时间失败会变成永不结束的后台工作。示例用人工时钟推进时间，测试无需真正等待。生产环境还应处理时钟偏差，并尽量以共享数据库或服务端时间裁决租约。
只重试明确的暂时性失败，示例使用固定退避便于断言；生产可采用封顶指数退避并加入抖动，避免大量 worker 同步重试。内容冲突、身份不匹配、永久权限拒绝和无效结构直接停止。业务重试与发送重试分开计数；后者耗尽并不否定已经保存的业务结果。次数与截止任一达到边界，就留下可追踪的终态。
取消要写入持久状态，并在开始工作、续租和提交结果之前再次检查。真实宿主支持中断时还应发出对应的取消请求，再观察实际终态。取消成功也不能抹去已经发出的邮件、已经扣除的费用或已经修改的文件；这些需要独立补偿规则，不能靠再次调用处理函数猜测。
多 Agent 循环另外需要总预算。示例限制 hops，只有 task 类型能进入工作队列，结果消息不会自动变成新任务。生产协调器还应统一维护任务链的最大跳数、总调用次数、费用预算和终止条件；不要信任请求正文自报 hops 等于零。恶意来源可以不断换 ID，所以去重和跳数上限都不能代替入口限流与认证。
9. 把来源认证和人类授权留在可信边界
消息正文中的 sender 只是一个声明。示例通过独立 AuthContext 接收已经认证的主体，并要求它与允许来源及 envelope.sender 一致。这个类没有进行任何密码学认证；实际接入层必须先完成服务端鉴权，再构造上下文。把用户提交的 JSON 直接反序列化为 AuthContext，会把演示边界彻底破坏。
合法来源仍可能发送越权任务。一个 Agent 可以建议发布、付款或给客户发消息，但它的建议不会自动获得用户授权。应用应把授权保存为独立记录，限定动作、收件人或资源、可传输数据、范围和有效期；在真正执行副作用之前再次核对。外部内容里的 human_approved 字段、签名正确的正文或可信同事的转述，都不能单独代替这条记录。
示例只把虚构任务转换成模拟结果，不实现真实授权服务，也不包含任何秘密。接入实际系统时应给执行器最小可用能力，审查任务使用只读权限，发布器持有单独受控权限；拒绝后不换身份、换接口或换执行器绕行。日志只记录排障所需的身份与状态，避免保存凭据和无关私人内容。
图 07 / 离线架构示意
查看图示文字说明
外部消息经可信接入层认证之后进入受限队列。人类授权记录独立于消息正文，并在副作用执行之前校验。正文中的批准字样不能代替授权。
10. 先核对宿主怎样开始下一轮执行
截至 2026 年 10 月 5 日，Grok Build 官方 headless 与 ACP 文档支持脚本调用、会话恢复和 session/prompt；Background Tasks 文档另有后台任务、循环提示与监控器说明。这些能力可以作为适配器的候选入口，但上述页面没有给出任意外部消息在进程退出后自动唤醒、保留全部未处理事件并恢复业务效果的通用保证。需要对具体版本做退出、恢复和取消测试。
Codex App Server 文档区分 thread/resume、turn/start 和 turn/completed。恢复线程让后续轮次接上已有历史；调用 turn/start 才开始新的生成；完成通知还要检查最终状态。自建适配器负责重启进程、恢复数据库任务并核对运行结果。不能把 WebSocket 连通或输入被接受当成任务完成。
官方 定时任务说明还区分运行表面：依赖本地文件的任务需要电脑开机且应用运行，网页任务不能直接使用本地目录。MCP Events、应用事件触发与 CLI 是不同入口，应按各自文档核对计划、工作区与权限条件，不将一个入口的能力推广到所有宿主。
Muse 方面，本次可核实的 Muse Gadgets 页面描述设备 SDK 及硬件连接。仅依据这些资料，尚不能确认用于本文工作流的任意 Agent 消息触发、退出后的自动恢复或通用后台回复契约。因此本文不给 Muse 生成可直接投产的唤醒配置。连接传输、创建执行轮次、持久排队、获取终态，应逐项取得证据。
11. 运行完整离线示例并检查数据库证据
新建 examples 目录，将文末的 recovery_queue.py、demo.py 和 test_recovery_queue.py 按文件名完整保存进去，再进入该目录。使用带 sqlite3 的 Python 3.10 或更新版本即可，无需安装依赖。先运行单测，再运行 demo。两者只在临时目录建立 SQLite 文件，演示结束后自动清理；不会读取本机模型会话、调用命令行 Agent 或对外发送消息。
demo 依次展示接收入队、worker 崩溃后的租约回收、有效 token 完成任务，以及远端提交成功但 ACK 丢失后的稳定 ID 重投。读输出时，把 task state、outbox state、发送尝试次数和接收端 effect_count 放在一起看。效果记录只有一条，才说明示例规定的接收端幂等断言成立。
完整源码和测试在文末展开区。recovery_queue.py 是队列和接收端；demo.py 是场景入口；test_recovery_queue.py 是故障矩阵。可把确定性处理器换成宿主适配器，但这一步需要补充授权、结果验证、模型成本、真实取消、外部幂等和进程生命周期测试。
离线运行 · examples/
python3 -m unittest -v
python3 demo.pydemo.py · 真实离线运行输出
Input ACK: stored state: pending
First claim: worker-1 fence: 1
Recovered claim: worker-2 fence: 2
Stale worker commit: rejected
Business state: done
First send: retry
Remote effects after lost ACK: 1
Stable reply ID: reply:bd09939f3bdb800b2ac17a92671ec9f53bb9a79f3c6ce81f465ea21257f41836
Reused reply ID: True
Reply state: sent receiver effects: 1
12. 按故障窗口验收 再接入真实执行器
测试应证明持久证据与恢复动作一致。本文覆盖提交前抛错的完整回滚、提交后重开数据库的保留、真实子进程提前退出、重复补收、部分批次成功、两个独立连接竞争任务、到期重领、旧 token 拒绝、取消与截止、重试耗尽，以及远端提交后回执丢失。来源伪造、同 ID 不同内容和循环边界也应在进入有效处理之前被发现。
验收记录必须区分通过、失败与没有执行。本文只验证本地代码、生成文件和可复现的离线故障；未连接任何真实模型宿主，未证明网络回放或生产服务达到某种可用性目标。真实上线前，还需验证目标工具的幂等保留期、权限撤销后的行为、恢复时队列积压和人工处理流程。
排障时保留 task_id、reply_id、owner、fence、attempt、当前状态、时间边界和最后错误。先定位失败发生在哪次提交之前或之后，再选择恢复动作。若状态证据不足，先暂停有副作用的步骤并核对目标系统。可靠恢复的基础是能够解释已经发生的动作，并让下一次尝试继续遵守原任务边界。
图 08 / 离线架构示意
查看图示文字说明
测试矩阵覆盖事务前后崩溃、远端提交回执丢失、重复补收、批次失败、并发认领、到期重领、旧令牌、终止边界与来源伪造。
- 36 项离线 unittest 通过，两次独立复测均为零失败
- 两个独立连接竞争同一任务，重复 100 次，零失败和零错误
- 独立子进程 os._exit 强退复测覆盖 6 个提交窗口，均恢复为一条接收效果记录
- demo 最终为 done / sent，接收端模拟效果数量为 1
- 8 幅 SVG 与 PNG 已生成，PNG 已目视检查；图与源码均内嵌 HTML
- 本地 HTML 与手机视口未作浏览器验收；公开页面已在 1180×757 桌面视口核验，八图加载正常
- 未运行真实模型、真实外部投递、宿主自动唤醒、生产负载或物理断电测试
13. 参考资料与核查范围
技术资料核对于 2026 年 10 月 5 日。以下均为协议项目或产品官方主源；文中架构与示例是独立教学设计。CodexApp Guide 由 pan 独立维护。
- SQLite transaction control：事务模式与写入竞争
- SQLite WAL：WAL 的并发、持久性与部署限制
- AWS transactional outbox pattern：双写窗口与重复事件
- Amazon SQS visibility timeout：可见性超时与重复投递
- Stripe idempotent requests：幂等键保留期与参数冲突
- OpenAI MCP Events：当前集成支持与收件确认
- MCP Events draft design：草案语义与宿主支持范围应分别核对
- Grok Build headless and scripting：headless 与 ACP 入口
- Grok Build background tasks：后台任务、循环与监控
- Codex App Server：线程恢复、轮次启动与完成状态
- Scheduled tasks：运行表面与本地环境要求
- Muse Gadgets：仅据此确认设备 SDK 资料，未确认通用唤醒契约
14. 完整源码与测试
下面按文件展开全部 Python 源码。示例不含省略号占位，不依赖外部服务。可按前文命令运行，并保存自己的输出。
recovery_queue.py · 完整源码
recovery_queue.py
"""Offline, standard-library-only at-least-once task and reply delivery lab.
AuthContext must come from a trusted, already-authenticated transport adapter.
Constructing it is NOT authentication. This module opens no network connection.
"""
from contextlib import contextmanager
from dataclasses import asdict, dataclass
import hashlib
import json
import math
import sqlite3
class Crash(RuntimeError):
"""Fault injection: abandon the call as if the process stopped."""
class AckLost(RuntimeError):
pass
class Unauthorized(ValueError):
pass
class PayloadCollision(ValueError):
pass
class Fenced(RuntimeError):
pass
class BatchFailed(RuntimeError):
def __init__(self, acks, index, cause):
super().__init__(f"batch stopped at index {index}: {cause}")
self.acks, self.index, self.cause = acks, index, cause
@dataclass(frozen=True)
class AuthContext:
# A trusted adapter supplies this value AFTER authenticating the connection.
principal: str
@dataclass(frozen=True)
class Envelope:
message_id: str
sender: str
payload: dict
deadline: float
max_attempts: int = 3
hops: int = 0
kind: str = "task"
@dataclass(frozen=True)
class Claim:
task_id: str
owner: str
fence: int
attempt: int
payload: dict
class ManualClock:
def __init__(self, now=1000.0):
self.now = float(now)
def __call__(self):
return self.now
def advance(self, seconds):
if seconds < 0:
raise ValueError("clock cannot move backwards")
self.now += seconds
def encode(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False)
def digest(text):
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def fault(point, requested):
if point == requested:
raise Crash(point)
def finite_number(value):
try:
return type(value) in (int, float) and math.isfinite(value)
except OverflowError:
return False
def valid_json(value):
stack, count = [(value, 0)], 0
while stack:
item, depth = stack.pop()
count += 1
if depth > 16 or count > 10000:
return False
if isinstance(item, dict):
if not all(isinstance(key, str) for key in item):
return False
stack.extend((child, depth + 1) for child in item.values())
elif isinstance(item, list):
stack.extend((child, depth + 1) for child in item)
elif item is not None and type(item) not in (str, bool):
if not finite_number(item):
return False
return True
def connect(path):
db = sqlite3.connect(path, timeout=10, isolation_level=None)
db.row_factory = sqlite3.Row
db.execute("PRAGMA busy_timeout=10000")
db.execute("PRAGMA foreign_keys=ON")
# Initialize once before starting workers. Avoid re-setting WAL per worker.
if db.execute("PRAGMA journal_mode").fetchone()[0] != "wal":
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
return db
@contextmanager
def transaction(db):
db.execute("BEGIN IMMEDIATE")
try:
yield
db.execute("COMMIT")
except BaseException:
if db.in_transaction:
db.execute("ROLLBACK")
raise
SCHEMA = """
CREATE TABLE IF NOT EXISTS inbox (
principal TEXT NOT NULL, message_id TEXT NOT NULL,
content_hash TEXT NOT NULL, content TEXT NOT NULL,
task_id TEXT NOT NULL UNIQUE, received_at REAL NOT NULL,
PRIMARY KEY (principal, message_id)
);
CREATE TABLE IF NOT EXISTS tasks (
task_id TEXT PRIMARY KEY REFERENCES inbox(task_id),
payload TEXT NOT NULL, deadline REAL NOT NULL, hops INTEGER NOT NULL,
state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
max_attempts INTEGER NOT NULL, available_at REAL NOT NULL,
owner TEXT, lease_until REAL, fence INTEGER NOT NULL DEFAULT 0,
result TEXT
);
CREATE TABLE IF NOT EXISTS outbox (
reply_id TEXT PRIMARY KEY, task_id TEXT NOT NULL UNIQUE,
payload TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0, available_at REAL NOT NULL,
owner TEXT, lease_until REAL, fence INTEGER NOT NULL DEFAULT 0,
last_error TEXT
);
"""
class Queue:
def __init__(self, path, clock, allowed_sources=("agent-a",),
lease=10, backoff=2, max_hops=4, delivery_attempts=3):
if (not finite_number(lease) or lease <= 0
or not finite_number(backoff) or backoff < 0
or type(max_hops) is not int or not 1 <= max_hops <= 1024
or type(delivery_attempts) is not int
or not 1 <= delivery_attempts <= 100):
raise ValueError("invalid queue configuration")
self.db = connect(path)
self.db.executescript(SCHEMA)
self.clock = clock
self.allowed_sources = frozenset(allowed_sources)
self.lease, self.backoff = lease, backoff
self.max_hops, self.delivery_attempts = max_hops, delivery_attempts
def close(self):
self.db.close()
def receive(self, auth, envelope, crash=None):
if not isinstance(envelope, Envelope):
raise ValueError("expected a parsed task envelope")
if (not isinstance(auth, AuthContext)
or not isinstance(auth.principal, str)
or auth.principal not in self.allowed_sources
or envelope.sender != auth.principal):
raise Unauthorized("source must match the authenticated principal")
if (not isinstance(envelope.message_id, str)
or not 1 <= len(envelope.message_id.encode("utf-8")) <= 128
or envelope.kind != "task"
or not finite_number(envelope.deadline)
or not isinstance(envelope.payload, dict)
or type(envelope.max_attempts) is not int
or not 1 <= envelope.max_attempts <= 10
or type(envelope.hops) is not int
or not 0 <= envelope.hops <= 2147483647
or not valid_json(envelope.payload)):
raise ValueError("invalid task envelope")
if len(encode(envelope.payload).encode("utf-8")) > 65536:
raise ValueError("payload exceeds 64 KiB")
content = encode(asdict(envelope))
content_hash = digest(content)
task_id = digest(encode([auth.principal, envelope.message_id]))
now = self.clock()
with transaction(self.db):
old = self.db.execute(
"SELECT content_hash FROM inbox WHERE principal=? "
"AND message_id=?", (auth.principal, envelope.message_id)
).fetchone()
if old and old["content_hash"] != content_hash:
raise PayloadCollision("same source and ID, different content")
if not old:
self.db.execute(
"INSERT INTO inbox VALUES (?, ?, ?, ?, ?, ?)",
(auth.principal, envelope.message_id, content_hash,
content, task_id, now))
self.db.execute(
"INSERT INTO tasks (task_id,payload,deadline,hops,state,"
"max_attempts,available_at) VALUES (?,?,?,?,'pending',?,?)",
(task_id, encode(envelope.payload), envelope.deadline,
envelope.hops, envelope.max_attempts, now))
if envelope.deadline <= now:
self._terminal(task_id, "expired", "deadline reached")
elif envelope.hops >= self.max_hops:
self._terminal(task_id, "hop_limit", "hop budget reached")
fault("before_commit", crash)
fault("after_commit", crash)
# This ACK says only that the immutable input is durably recorded.
return {"ack": "stored", "task_id": task_id, "duplicate": bool(old)}
def receive_batch(self, auth, envelopes, fault_at=None, crash=None):
acks = []
for index, envelope in enumerate(envelopes):
try:
acks.append(self.receive(
auth, envelope, crash if index == fault_at else None))
except Exception as exc:
raise BatchFailed(acks, index, exc) from exc
return acks
def _terminal(self, task_id, state, result):
# Call only inside a transaction. One terminal result per logical task.
body = encode({"kind": "result", "task_id": task_id,
"status": state, "result": result})
self.db.execute(
"UPDATE tasks SET state=?,result=?,owner=NULL,lease_until=NULL "
"WHERE task_id=?", (state, encode(result), task_id))
self.db.execute(
"INSERT INTO outbox (reply_id,task_id,payload,available_at) "
"VALUES (?,?,?,?)", ("reply:" + task_id, task_id, body, self.clock()))
def _reconcile(self):
now = self.clock()
rows = self.db.execute(
"SELECT * FROM tasks WHERE state IN ('pending','running')"
).fetchall()
for row in rows:
if row["deadline"] <= now:
self._terminal(row["task_id"], "expired", "deadline reached")
elif row["state"] == "running" and row["lease_until"] <= now:
if row["attempts"] >= row["max_attempts"]:
self._terminal(row["task_id"], "failed", "attempts exhausted")
else:
self.db.execute(
"UPDATE tasks SET state='pending',owner=NULL,"
"lease_until=NULL,available_at=? WHERE task_id=?",
(now, row["task_id"]))
def reconcile(self):
with transaction(self.db):
self._reconcile()
def claim(self, owner, crash=None):
with transaction(self.db):
now = self.clock()
self._reconcile()
row = self.db.execute(
"SELECT * FROM tasks WHERE state='pending' AND available_at<=? "
"ORDER BY available_at,task_id LIMIT 1", (now,)
).fetchone()
if row is None:
return None
self.db.execute(
"UPDATE tasks SET state='running',owner=?,lease_until=?,"
"fence=fence+1,attempts=attempts+1 WHERE task_id=?",
(owner, now + self.lease, row["task_id"]))
claim = Claim(row["task_id"], owner, row["fence"] + 1,
row["attempts"] + 1, json.loads(row["payload"]))
fault("before_commit", crash)
fault("after_commit", crash)
return claim
def _owned(self, claim):
row = self.db.execute(
"SELECT * FROM tasks WHERE task_id=?", (claim.task_id,)
).fetchone()
now = self.clock()
if (not row or row["state"] != "running"
or row["owner"] != claim.owner or row["fence"] != claim.fence
or row["lease_until"] <= now or row["deadline"] <= now):
raise Fenced("stale, expired, cancelled, or replaced claim")
return row
def renew(self, claim):
with transaction(self.db):
self._owned(claim)
self.db.execute(
"UPDATE tasks SET lease_until=? WHERE task_id=?",
(self.clock() + self.lease, claim.task_id))
def finish(self, claim, result, crash=None):
with transaction(self.db):
self._owned(claim)
self._terminal(claim.task_id, "done", result)
fault("before_commit", crash)
fault("after_commit", crash)
def fail(self, claim, reason):
with transaction(self.db):
row = self._owned(claim)
if row["attempts"] >= row["max_attempts"]:
self._terminal(claim.task_id, "failed", reason)
else:
self.db.execute(
"UPDATE tasks SET state='pending',owner=NULL,"
"lease_until=NULL,available_at=? WHERE task_id=?",
(self.clock() + self.backoff, claim.task_id))
def cancel(self, task_id):
# Internal trusted scheduler API, not an unauthenticated RPC endpoint.
with transaction(self.db):
row = self.db.execute(
"SELECT state FROM tasks WHERE task_id=?", (task_id,)
).fetchone()
if not row or row["state"] not in ("pending", "running"):
return False
self._terminal(task_id, "cancelled", "cancelled by scheduler")
return True
def task(self, task_id):
row = self.db.execute(
"SELECT * FROM tasks WHERE task_id=?", (task_id,)
).fetchone()
return dict(row) if row else None
def replies(self):
return [dict(row) for row in self.db.execute(
"SELECT * FROM outbox ORDER BY reply_id")]
def _claim_reply(self, owner):
with transaction(self.db):
now = self.clock()
self.db.execute(
"UPDATE outbox SET state=CASE WHEN attempts>=? THEN 'dead' "
"ELSE 'pending' END,owner=NULL,lease_until=NULL,"
"last_error='delivery lease expired; outcome unknown' "
"WHERE state='sending' AND lease_until<=?",
(self.delivery_attempts, now))
row = self.db.execute(
"SELECT * FROM outbox WHERE state='pending' AND available_at<=? "
"ORDER BY available_at,reply_id LIMIT 1", (now,)
).fetchone()
if not row:
return None
self.db.execute(
"UPDATE outbox SET state='sending',owner=?,lease_until=?,"
"fence=fence+1,attempts=attempts+1 WHERE reply_id=?",
(owner, now + self.lease, row["reply_id"]))
return dict(self.db.execute(
"SELECT * FROM outbox WHERE reply_id=?", (row["reply_id"],)
).fetchone())
def _reply_transition(self, item, delivered, crash=None, error=None,
permanent=False):
with transaction(self.db):
now = self.clock()
row = self.db.execute(
"SELECT * FROM outbox WHERE reply_id=?", (item["reply_id"],)
).fetchone()
if (row["state"] != "sending" or row["owner"] != item["owner"]
or row["fence"] != item["fence"]
or row["lease_until"] <= now):
raise Fenced("stale reply publisher")
exhausted = row["attempts"] >= self.delivery_attempts
state = "sent" if delivered else (
"dead" if permanent or exhausted else "pending")
self.db.execute(
"UPDATE outbox SET state=?,owner=NULL,lease_until=NULL,"
"available_at=?,last_error=? WHERE reply_id=?",
(state, now + self.backoff, error, item["reply_id"]))
fault("before_commit", crash)
fault("after_commit", crash)
return state
def dispatch_once(self, receiver, owner="publisher", crash=None):
item = self._claim_reply(owner)
if item is None:
return None
try:
receiver.accept(item["reply_id"], item["payload"], crash=crash)
except AckLost as exc:
state = self._reply_transition(item, delivered=False, error=str(exc))
return "dead" if state == "dead" else "retry"
except PayloadCollision as exc:
self._reply_transition(item, delivered=False, error=str(exc),
permanent=True)
return "dead"
# Crashes deliberately bypass cleanup and leave a reclaimable lease.
fault("after_remote_commit", crash)
local_crash = {
"before_sent_commit": "before_commit",
"after_sent_commit": "after_commit",
}.get(crash)
self._reply_transition(item, delivered=True, crash=local_crash)
return item["reply_id"]
class Receiver:
"""Independent DB: dedupe and its simulated business effect commit together."""
def __init__(self, path):
self.db = connect(path)
self.db.executescript("""
CREATE TABLE IF NOT EXISTS receipts (
reply_id TEXT PRIMARY KEY, content_hash TEXT NOT NULL,
payload TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS effects (
reply_id TEXT PRIMARY KEY REFERENCES receipts(reply_id),
payload TEXT NOT NULL
);
""")
def close(self):
self.db.close()
def accept(self, reply_id, payload, crash=None):
content_hash = digest(payload)
with transaction(self.db):
old = self.db.execute(
"SELECT content_hash FROM receipts WHERE reply_id=?", (reply_id,)
).fetchone()
if old and old["content_hash"] != content_hash:
raise PayloadCollision("reply ID reused with different content")
if not old:
self.db.execute("INSERT INTO receipts VALUES (?,?,?)",
(reply_id, content_hash, payload))
self.db.execute("INSERT INTO effects VALUES (?,?)",
(reply_id, payload))
fault("before_remote_commit", crash)
if crash == "lost_ack":
raise AckLost("receiver committed; sender did not observe the ACK")
return {"ack": "stored", "duplicate": bool(old)}
def effect_count(self):
return self.db.execute("SELECT COUNT(*) FROM effects").fetchone()[0]
demo.py · 完整源码
demo.py
"""Offline demo: lease recovery, fencing, lost ACK, and idempotent replay."""
from pathlib import Path
import tempfile
from recovery_queue import (
AuthContext, Envelope, Fenced, ManualClock, Queue, Receiver,
)
def demo():
with tempfile.TemporaryDirectory() as folder:
clock = ManualClock()
sender_path = Path(folder) / "sender.db"
queue = Queue(sender_path, clock)
receiver = Receiver(Path(folder) / "receiver.db")
envelope = Envelope("demo-1", "agent-a", {"numbers": [2, 3]}, 1100)
ack = queue.receive(AuthContext("agent-a"), envelope)
print("Input ACK:", ack["ack"], "state:",
queue.task(ack["task_id"])["state"])
old = queue.claim("worker-1")
print("First claim:", old.owner, "fence:", old.fence)
# Simulate a disappeared worker, restart the queue, and expire its lease.
queue.close()
clock.advance(10)
queue = Queue(sender_path, clock)
new = queue.claim("worker-2")
print("Recovered claim:", new.owner, "fence:", new.fence)
try:
queue.finish(old, {"sum": 999})
except Fenced:
print("Stale worker commit: rejected")
queue.finish(new, {"sum": sum(new.payload["numbers"])})
print("Business state:", queue.task(new.task_id)["state"])
reply_id = queue.replies()[0]["reply_id"]
print("First send:", queue.dispatch_once(receiver, crash="lost_ack"))
print("Remote effects after lost ACK:", receiver.effect_count())
clock.advance(2)
delivered_id = queue.dispatch_once(receiver)
print("Stable reply ID:", delivered_id)
print("Reused reply ID:", delivered_id == reply_id)
print("Reply state:", queue.replies()[0]["state"],
"receiver effects:", receiver.effect_count())
queue.close()
receiver.close()
if __name__ == "__main__":
demo()
test_recovery_queue.py · 完整源码
test_recovery_queue.py
"""Deterministic fault tests; no real models, network, credentials, or sleep."""
from dataclasses import replace
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import unittest
from recovery_queue import (
AuthContext, BatchFailed, Crash, Envelope, Fenced, ManualClock,
PayloadCollision, Queue, Receiver, Unauthorized,
)
class RecoveryTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory()
self.root = Path(self.directory.name)
self.sender_path = self.root / "sender.db"
self.receiver_path = self.root / "receiver.db"
self.clock = ManualClock()
self.queue = Queue(self.sender_path, self.clock)
self.receiver = Receiver(self.receiver_path)
self.auth = AuthContext("agent-a")
def tearDown(self):
self.queue.close()
self.receiver.close()
self.directory.cleanup()
def envelope(self, message_id="m1", **changes):
return replace(Envelope(message_id, "agent-a", {"numbers": [2, 3]},
self.clock() + 100), **changes)
def receive(self, **changes):
return self.queue.receive(self.auth, self.envelope(**changes))["task_id"]
def finish_one(self):
task_id = self.receive()
claim = self.queue.claim("worker-a")
self.queue.finish(claim, {"sum": sum(claim.payload["numbers"])})
return task_id
def restart(self):
self.queue.close()
self.queue = Queue(self.sender_path, self.clock)
def test_input_ack_is_not_business_completion(self):
ack = self.queue.receive(self.auth, self.envelope())
self.assertEqual(ack["ack"], "stored")
self.assertEqual(self.queue.task(ack["task_id"])["state"], "pending")
self.assertEqual(self.queue.replies(), [])
claim = self.queue.claim("worker")
self.queue.finish(claim, {"sum": 5})
self.assertEqual(self.queue.task(claim.task_id)["state"], "done")
self.assertEqual(self.queue.replies()[0]["state"], "pending")
self.assertEqual(self.receiver.effect_count(), 0)
def test_input_crash_before_commit_rolls_back_everything(self):
envelope = self.envelope()
with self.assertRaises(Crash):
self.queue.receive(self.auth, envelope, crash="before_commit")
self.restart()
self.assertIsNone(self.queue.claim("worker"))
ack = self.queue.receive(self.auth, envelope)
self.assertFalse(ack["duplicate"])
self.assertIsNotNone(self.queue.claim("worker"))
def test_input_crash_after_commit_replay_is_duplicate(self):
envelope = self.envelope()
with self.assertRaises(Crash):
self.queue.receive(self.auth, envelope, crash="after_commit")
self.restart()
ack = self.queue.receive(self.auth, envelope)
self.assertTrue(ack["duplicate"])
self.assertIsNotNone(self.queue.claim("worker"))
self.assertIsNone(self.queue.claim("other"))
def test_reconnect_can_replay_entire_batch(self):
batch = [self.envelope(str(index)) for index in range(4)]
first = self.queue.receive_batch(self.auth, batch)
self.restart()
again = self.queue.receive_batch(self.auth, batch)
self.assertTrue(all(ack["duplicate"] for ack in again))
self.assertEqual([ack["task_id"] for ack in first],
[ack["task_id"] for ack in again])
self.assertEqual(self.queue.db.execute(
"SELECT COUNT(*) FROM tasks").fetchone()[0], 4)
def test_partial_batch_before_commit_keeps_only_committed_prefix(self):
batch = [self.envelope(str(index)) for index in range(3)]
with self.assertRaises(BatchFailed) as raised:
self.queue.receive_batch(self.auth, batch, fault_at=1,
crash="before_commit")
error = raised.exception
self.assertEqual(error.index, 1)
self.assertEqual(len(error.acks), 1)
self.restart()
replay = self.queue.receive_batch(self.auth, batch)
self.assertEqual([ack["duplicate"] for ack in replay],
[True, False, False])
def test_partial_batch_after_commit_has_ambiguous_unacked_item(self):
batch = [self.envelope(str(index)) for index in range(3)]
with self.assertRaises(BatchFailed) as raised:
self.queue.receive_batch(self.auth, batch, fault_at=1,
crash="after_commit")
self.assertEqual(len(raised.exception.acks), 1)
self.restart()
replay = self.queue.receive_batch(self.auth, batch)
self.assertEqual([ack["duplicate"] for ack in replay],
[True, True, False])
def test_same_payload_id_collision_does_not_overwrite(self):
envelope = self.envelope()
ack = self.queue.receive(self.auth, envelope)
with self.assertRaises(PayloadCollision):
self.queue.receive(self.auth, replace(envelope, payload={"sum": 99}))
self.assertEqual(json.loads(self.queue.task(ack["task_id"])["payload"]),
{"numbers": [2, 3]})
def test_spoofed_payload_sender_cannot_supply_authentication(self):
with self.assertRaises(Unauthorized):
self.queue.receive(AuthContext("mallory"), self.envelope())
with self.assertRaises(Unauthorized):
self.queue.receive(self.auth, self.envelope(sender="agent-b"))
with self.assertRaises(Unauthorized):
self.queue.receive({"principal": "agent-a"}, self.envelope())
self.assertIsNone(self.queue.claim("worker"))
def test_invalid_envelope_boundaries_are_rejected_without_writes(self):
cases = [
{"message_id": ""}, {"message_id": 123},
{"message_id": "中" * 43}, {"message_id": "x" * 129},
{"deadline": "1100"}, {"deadline": True},
{"deadline": float("nan")}, {"deadline": float("inf")},
{"deadline": float("-inf")}, {"deadline": 10 ** 1000},
{"payload": {"large": "x" * 65536}},
{"payload": {"bad": float("nan")}}, {"payload": {1: "bad key"}},
{"payload": {"tuple": (1, 2)}}, {"payload": []},
{"hops": -1}, {"hops": 1.5}, {"hops": True},
{"hops": 2147483648}, {"max_attempts": 0},
{"max_attempts": 11}, {"max_attempts": True},
]
for changes in cases:
with self.subTest(changes=changes):
with self.assertRaises(ValueError):
self.queue.receive(self.auth, self.envelope(**changes))
recursive = {}
recursive["self"] = recursive
with self.assertRaises(ValueError):
self.queue.receive(self.auth, self.envelope(payload=recursive))
self.assertIsNone(self.queue.claim("worker"))
self.assertEqual(self.queue.db.execute(
"SELECT COUNT(*) FROM inbox").fetchone()[0], 0)
def test_boundary_values_and_large_hop_count_are_handled(self):
ack = self.queue.receive(self.auth, self.envelope(message_id="x" * 128))
self.assertEqual(self.queue.task(ack["task_id"])["state"], "pending")
task_id = self.receive(message_id="huge-hop", hops=2147483647)
self.assertEqual(self.queue.task(task_id)["state"], "hop_limit")
def test_existing_wal_open_is_safe_while_another_connection_writes(self):
self.queue.db.execute("BEGIN IMMEDIATE")
# Opening another connection queries existing WAL rather than toggling it.
from recovery_queue import connect
other = connect(self.sender_path)
try:
self.assertEqual(other.execute("PRAGMA journal_mode").fetchone()[0],
"wal")
self.assertEqual(other.execute("PRAGMA busy_timeout").fetchone()[0],
10000)
finally:
other.close()
self.queue.db.execute("ROLLBACK")
def test_same_message_id_from_distinct_authorized_sources_is_distinct(self):
self.queue.close()
self.queue = Queue(self.sender_path, self.clock,
allowed_sources=("agent-a", "agent-b"))
first = self.receive()
second = self.queue.receive(AuthContext("agent-b"),
self.envelope(sender="agent-b"))["task_id"]
self.assertNotEqual(first, second)
def test_two_workers_race_with_independent_connections(self):
self.receive()
barrier = threading.Barrier(2)
results, errors = [], []
lock = threading.Lock()
def worker(owner):
queue = None
try:
queue = Queue(self.sender_path, self.clock)
barrier.wait(timeout=10)
claim = queue.claim(owner)
with lock:
results.append(claim)
except BaseException as exc:
with lock:
errors.append(exc)
finally:
if queue is not None:
queue.close()
threads = [threading.Thread(target=worker, args=(name,))
for name in ("worker-a", "worker-b")]
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=15)
self.assertTrue(all(not thread.is_alive() for thread in threads))
self.assertEqual(errors, [])
self.assertEqual(len(results), 2)
self.assertEqual(sum(result is not None for result in results), 1)
def test_claim_crash_before_commit_does_not_consume_attempt(self):
self.receive()
with self.assertRaises(Crash):
self.queue.claim("worker-a", crash="before_commit")
self.restart()
claim = self.queue.claim("worker-b")
self.assertEqual((claim.fence, claim.attempt), (1, 1))
def test_claim_crash_after_commit_recovers_after_lease(self):
self.receive()
with self.assertRaises(Crash):
self.queue.claim("worker-a", crash="after_commit")
self.restart()
self.assertIsNone(self.queue.claim("worker-b"))
self.clock.advance(10)
claim = self.queue.claim("worker-b")
self.assertEqual((claim.fence, claim.attempt), (2, 2))
def test_expired_lease_is_fenced_even_before_reclaim(self):
self.receive()
old = self.queue.claim("worker-a")
self.clock.advance(10)
with self.assertRaises(Fenced):
self.queue.finish(old, "late")
with self.assertRaises(Fenced):
self.queue.renew(old)
def test_reclaimed_task_rejects_old_worker(self):
task_id = self.receive()
old = self.queue.claim("worker-a")
self.clock.advance(10)
new = self.queue.claim("worker-b")
self.assertGreater(new.fence, old.fence)
with self.assertRaises(Fenced):
self.queue.finish(old, "wrong")
with self.assertRaises(Fenced):
self.queue.fail(old, "wrong")
self.queue.finish(new, "right")
self.assertEqual(json.loads(self.queue.task(task_id)["result"]), "right")
self.assertEqual(len(self.queue.replies()), 1)
def test_renew_preserves_exclusive_valid_lease(self):
self.receive()
claim = self.queue.claim("worker-a")
self.clock.advance(8)
self.queue.renew(claim)
self.clock.advance(8)
self.assertIsNone(self.queue.claim("worker-b"))
self.queue.finish(claim, "done")
def test_finish_crash_before_commit_rolls_back_result_and_outbox(self):
task_id = self.receive()
claim = self.queue.claim("worker-a")
with self.assertRaises(Crash):
self.queue.finish(claim, "first", crash="before_commit")
self.restart()
self.assertEqual(self.queue.task(task_id)["state"], "running")
self.assertEqual(self.queue.replies(), [])
self.clock.advance(10)
self.queue.finish(self.queue.claim("worker-b"), "recomputed")
self.assertEqual(len(self.queue.replies()), 1)
def test_finish_crash_after_commit_keeps_result_and_outbox(self):
task_id = self.receive()
claim = self.queue.claim("worker-a")
with self.assertRaises(Crash):
self.queue.finish(claim, "done", crash="after_commit")
self.restart()
self.assertEqual(self.queue.task(task_id)["state"], "done")
self.assertEqual(len(self.queue.replies()), 1)
with self.assertRaises(Fenced):
self.queue.finish(claim, "different")
def test_retry_budget_counts_claims_and_respects_backoff(self):
task_id = self.receive(max_attempts=2)
first = self.queue.claim("worker")
self.queue.fail(first, "temporary")
self.assertIsNone(self.queue.claim("worker"))
self.clock.advance(2)
second = self.queue.claim("worker")
self.assertEqual(second.attempt, 2)
self.queue.fail(second, "permanent")
self.assertEqual(self.queue.task(task_id)["state"], "failed")
self.assertIsNone(self.queue.claim("worker"))
self.assertEqual(len(self.queue.replies()), 1)
def test_repeated_worker_crashes_eventually_exhaust_attempts(self):
task_id = self.receive(max_attempts=1)
self.queue.claim("worker")
self.clock.advance(10)
self.assertIsNone(self.queue.claim("replacement"))
self.assertEqual(self.queue.task(task_id)["state"], "failed")
def test_deadline_at_ingestion_and_during_execution(self):
expired = self.receive(message_id="past", deadline=self.clock())
self.assertEqual(self.queue.task(expired)["state"], "expired")
running = self.receive(message_id="future", deadline=self.clock() + 3)
claim = self.queue.claim("worker")
self.clock.advance(3)
with self.assertRaises(Fenced):
self.queue.finish(claim, "too late")
self.queue.reconcile()
self.assertEqual(self.queue.task(running)["state"], "expired")
self.assertEqual(len(self.queue.replies()), 2)
def test_cancel_pending_running_and_already_finished(self):
pending = self.receive(message_id="pending")
self.assertTrue(self.queue.cancel(pending))
running = self.receive(message_id="running")
old = self.queue.claim("worker")
self.assertTrue(self.queue.cancel(running))
with self.assertRaises(Fenced):
self.queue.finish(old, "cancel was ignored")
self.assertFalse(self.queue.cancel(running))
done = self.receive(message_id="done")
self.queue.finish(self.queue.claim("worker"), "complete")
self.assertFalse(self.queue.cancel(done))
self.assertEqual(len(self.queue.replies()), 3)
def test_hop_limit_makes_terminal_result_without_executing(self):
task_id = self.receive(hops=4)
self.assertEqual(self.queue.task(task_id)["state"], "hop_limit")
self.assertIsNone(self.queue.claim("worker"))
self.assertEqual(json.loads(self.queue.replies()[0]["payload"])["kind"],
"result")
with self.assertRaises(ValueError):
self.queue.receive(self.auth, self.envelope(kind="result"))
allowed = self.receive(message_id="last-hop", hops=3)
self.assertEqual(self.queue.claim("worker").task_id, allowed)
def test_remote_commit_with_lost_ack_retries_same_reply_id(self):
self.finish_one()
original = self.queue.replies()[0]
self.assertEqual(self.queue.dispatch_once(self.receiver, crash="lost_ack"),
"retry")
self.assertEqual(self.receiver.effect_count(), 1)
self.assertEqual(self.queue.replies()[0]["state"], "pending")
self.clock.advance(2)
sent = self.queue.dispatch_once(self.receiver)
self.assertEqual(sent, original["reply_id"])
self.assertEqual(self.queue.replies()[0]["payload"], original["payload"])
self.assertEqual(self.queue.replies()[0]["state"], "sent")
self.assertEqual(self.receiver.effect_count(), 1)
def test_sender_crash_after_remote_commit_replays_after_lease(self):
self.finish_one()
with self.assertRaises(Crash):
self.queue.dispatch_once(self.receiver, crash="after_remote_commit")
self.restart()
self.assertEqual(self.receiver.effect_count(), 1)
self.assertIsNone(self.queue.dispatch_once(self.receiver))
self.clock.advance(10)
self.queue.dispatch_once(self.receiver)
self.assertEqual(self.receiver.effect_count(), 1)
self.assertEqual(self.queue.replies()[0]["state"], "sent")
def test_remote_crash_before_commit_rolls_back_effect(self):
self.finish_one()
with self.assertRaises(Crash):
self.queue.dispatch_once(self.receiver, crash="before_remote_commit")
self.assertEqual(self.receiver.effect_count(), 0)
self.clock.advance(10)
self.queue.dispatch_once(self.receiver)
self.assertEqual(self.receiver.effect_count(), 1)
def test_crash_before_local_sent_commit_is_safe_to_replay(self):
self.finish_one()
with self.assertRaises(Crash):
self.queue.dispatch_once(self.receiver, crash="before_sent_commit")
self.restart()
self.assertEqual(self.queue.replies()[0]["state"], "sending")
self.assertEqual(self.receiver.effect_count(), 1)
self.clock.advance(10)
self.queue.dispatch_once(self.receiver)
self.assertEqual(self.receiver.effect_count(), 1)
self.assertEqual(self.queue.replies()[0]["state"], "sent")
def test_crash_after_local_sent_commit_requires_no_replay(self):
self.finish_one()
with self.assertRaises(Crash):
self.queue.dispatch_once(self.receiver, crash="after_sent_commit")
self.restart()
self.assertEqual(self.queue.replies()[0]["state"], "sent")
self.assertIsNone(self.queue.dispatch_once(self.receiver))
self.assertEqual(self.receiver.effect_count(), 1)
def test_reply_payload_collision_is_rejected(self):
self.finish_one()
self.queue.dispatch_once(self.receiver)
reply = self.queue.replies()[0]
with self.assertRaises(PayloadCollision):
self.receiver.accept(reply["reply_id"], '{"changed":true}')
self.assertEqual(self.receiver.effect_count(), 1)
def test_permanent_reply_collision_stops_immediately_and_records_error(self):
self.finish_one()
reply = self.queue.replies()[0]
self.receiver.accept(reply["reply_id"], '{"different":"content"}')
self.assertEqual(self.queue.dispatch_once(self.receiver), "dead")
item = self.queue.replies()[0]
self.assertEqual(item["state"], "dead")
self.assertEqual(item["attempts"], 1)
self.assertIn("different content", item["last_error"])
self.clock.advance(100)
self.assertIsNone(self.queue.dispatch_once(self.receiver))
self.assertEqual(self.receiver.effect_count(), 1)
def test_delivery_crashes_exhaust_budget_with_unknown_remote_outcome(self):
self.finish_one()
for _ in range(3):
with self.assertRaises(Crash):
self.queue.dispatch_once(self.receiver,
crash="after_remote_commit")
self.clock.advance(10)
self.assertIsNone(self.queue.dispatch_once(self.receiver))
item = self.queue.replies()[0]
self.assertEqual(item["state"], "dead")
self.assertIn("outcome unknown", item["last_error"])
self.assertEqual(self.receiver.effect_count(), 1)
def test_delivery_budget_can_end_dead_despite_remote_success(self):
self.finish_one()
for _ in range(3):
self.queue.dispatch_once(self.receiver, crash="lost_ack")
self.clock.advance(2)
self.assertEqual(self.queue.replies()[0]["state"], "dead")
self.assertEqual(self.receiver.effect_count(), 1)
self.assertIsNone(self.queue.dispatch_once(self.receiver))
def test_reclaimed_outbox_rejects_stale_publisher(self):
self.finish_one()
old = self.queue._claim_reply("publisher-a")
self.clock.advance(10)
new = self.queue._claim_reply("publisher-b")
self.receiver.accept(old["reply_id"], old["payload"])
with self.assertRaises(Fenced):
self.queue._reply_transition(old, delivered=True)
self.receiver.accept(new["reply_id"], new["payload"])
self.queue._reply_transition(new, delivered=True)
self.assertEqual(self.receiver.effect_count(), 1)
def test_actual_process_exit_before_and_after_commit(self):
# os._exit bypasses Python finally/rollback: this is an actual abrupt exit.
program = """
import os
import sys
import recovery_queue as lab
queue = lab.Queue(sys.argv[1], lab.ManualClock())
point = sys.argv[2]
def terminate(where, requested):
if where == requested:
os._exit(73)
lab.fault = terminate
queue.receive(lab.AuthContext('agent-a'),
lab.Envelope('abrupt', 'agent-a', {'n': 1}, 1100), crash=point)
"""
for point, duplicate in [("before_commit", False),
("after_commit", True)]:
with self.subTest(point=point):
path = self.root / (point + ".db")
completed = subprocess.run(
[sys.executable, "-c", program, str(path), point],
cwd=Path(__file__).parent, capture_output=True, timeout=15)
self.assertEqual(completed.returncode, 73, completed.stderr)
reopened = Queue(path, self.clock)
try:
ack = reopened.receive(
self.auth, Envelope("abrupt", "agent-a", {"n": 1}, 1100))
self.assertEqual(ack["duplicate"], duplicate)
self.assertIsNotNone(reopened.claim("recovery-worker"))
self.assertIsNone(reopened.claim("other-worker"))
finally:
reopened.close()
if __name__ == "__main__":
unittest.main(verbosity=2)
保存聊天记录足够恢复多 Agent 工作吗？
还需要持久任务状态、稳定身份、租约、尝试次数和发送记录。会话历史可恢复上下文，不能单独确认业务是否已提交。
使用 SQLite 和 outbox 就能保证恰好一次吗？
不能无条件保证。示例的幂等覆盖同一接收端保存期内的稳定回复 ID，以及与去重记录同事务提交的模拟效果。外部服务需要自身的幂等或对账机制。
WebSocket 连上以后 Agent 就会自动回复吗？
连接只说明传输层状态。还要验证宿主的触发入口、执行轮次、权限、后台生命周期和最终结果。本文没有运行真实模型。
任务取消后还能收到结果吗？
取消与完成的先后由持久状态裁决。已经完成的任务不会被伪装成未执行，已发生的外部副作用也不会自动撤销。