Agent 把问题发出去,等人回一句话。人去开会了,服务却在这时候重启。几小时后,回复终于到了:界面告诉对方“已收到”,后台那个正在等待的协程,早已不在了。
这是本文构造的故障场景,不是真实用户事故。它提醒我们:让一个函数暂停,不等于把“等谁、等哪一轮、等到什么时候”保存下来。还有一种更难察觉的错误:旧问题的回信晚到了,恰好唤醒了新问题。消息没有丢,流程却接错了话。
第十五篇固定的是任务执行的定义,本篇作为 Agent 可靠执行实战第十六篇,处理长任务在外部消息与截止时间之间的完整等待生命周期:登记、拒收早到消息、接受有效信号、裁决超时、保存继续指令、重启恢复和再次等待。
随附项目不调用大模型,不发送邮件,不连接真实审批系统。回信是合成字符串,时间是可信的逻辑整数;真正执行的是 SQLite 事务、操作系统子进程退出与并发争用。它不把“能收到一个信号”当作真人身份验证,也不替代第八篇的精确意图批准。批准是否合法与回信是否交给了正确等待,是两层不同的问题。
等待要有地址,而且地址不能只写流程号#
示例给每次等待三个标识:flow 是流程,step 是等待步骤,generation 是这个步骤第几次等待。它们共同构成不可复用的等待地址。
例如 (工单甲, 补充材料, 1) 超时后,业务决定再问一次,就登记第 2 代。对方回复第 1 次问题时,消息仍然带第 1 代,不能因为工单还存在,就自动交给第 2 代。这里没有假设同一流程只能等一个问题;不同 step 可以各有自己的等待,但它们共享本例数据库的逻辑时钟,后文会说明这个限制。
slot 表只保留当前 generation,waiting 表则保留每一代的 deadline、状态及最终采用的 event_id。登记接口要求调用者给出预期的当前代次。首次登记传 0,成功后得到 1;两个登记者同时认为当前为 0,只会有一个成功。另一个读到的已是 1,返回 stale,而不是再创建一次问题。
新一代也不是随时可开。旧代仍是 waiting,返回 busy;旧代已终结但其继续指令还没消费,返回 pending_continuation。本例要求旧轮的本地继续效果先处理完,才能开始新轮,避免把“等待终结”与“下游已经推进”混成一个状态。
这个限制不是所有工作流的唯一设计。某些系统允许多个版本的后续任务并行,但就需要更明确的跨代隔离。本例选择较窄的串行契约,宁可让调用者看见一个待处理状态,也不悄悄盖掉前一轮记录。
登记请求还没有永久请求 ID 回执。若提交成功但响应丢失,用原 expected_generation 重试会得到 stale;调用者必须读取对应代次和等待字段确认结果。不能看见 stale 就立刻把 expected_generation 加一再试,那会把“确认上次登记”变成“主动再开一轮”。这是明确保留的接口边界,不是假装已经实现了所有命令的幂等。
早到的回信:拒收也必须记得住#
外部系统有时跑得很快:信号先来了,等待尚未登记。常见办法是暂存一封“无主邮件”,等等待对象出现后再投递。它可以成立,但需要解决暂存期限、归属验证、存储预算,以及同一个地址究竟何时允许再次使用。
这次示例不实现早到缓冲,选择更保守的契约:找不到精确等待地址时,写下 unknown_wait 回执并返回。拒收本身是一个持久结果。
随后即使等待登记成功,同一 event_id、同一消息体重试,仍重放原来的 unknown_wait。它不会因为环境变了,就从“未采用”变成“已经采用”。否则发送方第一次收到拒收,稍后由网络自动重试,业务却悄悄继续了,双方对这条消息的理解会分叉。
这不意味着早到消息永远无法重新提交。调用方可以在确认等待已建立后,经过明确的业务决策,用一个新的事件身份发起新的提交。但换 ID 不是网络重试的自动补救术:新事件意味着新的投递意图,是否允许重新提交要由上层协议决定,本例没有代替用户批准它。
更稳妥的生产路径通常是先原子保存等待及待发送请求,再由既有的 outbox 机制向外通知。本文不实现外发通道,也不声称改变发信顺序就能消除全部竞态。它只让“信号确实早到了”有一个可重放、可审查的结果,而不是一条容易消失的日志。
消息身份、消息内容和接收时间是三回事#
receipt 表以 event_id 为主键。在这个示例里,event_id 的唯一范围是整份数据库,不是某个租户、某个 flow 或某一代。消息体绑定 flow、step、generation、value 四个字段,并用确定的 JSON 序列化保存。
同 ID、同消息体,返回 replay 和原始结果;同 ID、不同 value,返回 conflict;同 ID 改成另一代或另一个流程,也返回 conflict。不能只检查“这个 ID 见过”,然后把来自不同业务目标的消息都当作安全重试。
接收时刻 now 不属于消息身份。发送方没有拿到响应,下一次重试当然可能晚几秒、几小时。只要正文完全相同,回执重放优先于当前等待状态和当前时间检查。第一次 accepted 的消息,后来即使已超过 deadline,仍得到原 accepted 回执,不能改判为迟到。
但 accepted 也不是后续工作全部完成。它只表示这条消息在本地事务里赢得了该代等待,并生成了持久继续指令。相应的 continuation 可能还没被消费,更不意味着某封邮件已经发出、某笔款项已经支付,或者一个真实人已经被认证。
输入校验只限制非空文本、UTF-8 字节长度、正代次及有界整数;它不是鉴权。若两个不可信租户能任意提供同一个 event_id,彼此就可能制造冲突。真实服务需要由可信入口确定身份范围、验证消息来源与授权,再进入这层协议,而不是把 JSON 解析成功当作安全边界。
信号与超时只能写出一个结局#
等待只有三个状态:waiting、signaled、timed_out。信号和超时扫描都使用 BEGIN IMMEDIATE 事务,先读取当前代次和状态,再决定能否终结。终态写入、继续指令插入,以及信号场景中的回执保存,位于同一个事务中。
SQLite 的事务文档说明,同一数据库可以有多个读事务,但同时只有一个写事务;IMMEDIATE 会立即尝试开始写事务,并可能遇到 SQLITE_BUSY。本文利用的是这个单库序列化边界,不是“分布式事务天然存在”。连接的等待超时只是便于复现小规模争用,锁错误依然会暴露,不能解释成业务超时已完成。
信号进来后,若当前代已换,记 stale_generation;若该代已终结,记 closed;若还是 waiting,却已满足 now >= deadline,就不能采用回信。此时信号处理本身会将等待终结为 timed_out,并生成超时继续指令,自己的回执为 too_late。
最后这条很重要:不能让“超时扫描器还没跑”变成“截止时间仍然有效”。定时任务晚执行,不应该给某条回信随机延长契约。恰好在 deadline 的信号也不接受,本例采用严格小于截止时间的规则。这个边界需要写进协议,不能依赖某台机器碰巧先调度谁。
如果有效信号先提交,后来的 expire 看到 closed,不再写另一个超时分支。如果超时先终结,旧时间戳的信号也不能把终态翻回来。状态、继续指令及本地效果的唯一键共同让不一致显露为错误,而不是事后从两个已运行的分支里挑一个“正确结果”。
时钟也有状态,不能把旧时间拿来洗白#
为了让测试不依赖真实等待,示例的 now 是调用方传入的逻辑整数。每条新计时命令会把数据库时钟更新为 max(已保存高水位, now),实际裁决使用这个结果。
例如某次操作已把高水位推到 12,另一个调用方随后拿着 2 来提交 deadline=10 的信号。若直接使用请求自带的 2,就会把已经过期的等待重新说成“还来得及”;高水位使这个请求仍按 12 处理,得到 too_late。
这只是一个可复现的单库逻辑时钟模型,不是可靠墙钟实现,更没有证明跨机器时间同步。高水位是全局共享的:新的 unknown_wait 信号、过期代次定时请求、被拒绝的登记请求,也可能推进时钟。已存在回执的 replay 和 conflict 直接返回,不推进时钟。代码和测试特意保留这些次序,避免读者以为“没有接受业务请求,就一定没有任何状态变化”。
因此 now 必须来自可信测试驱动或可信服务入口,绝不能直接让外部用户随意填写。一个极大的未来值,会让这份数据库里的其他等待一起过期。生产系统要明确服务端时间来源、时钟跳变处理、持久 timer 语义和运维校时策略;如果业务按“外部系统发送时刻”裁决,还需要另一个可验证的时间与传输契约。
我们测到的是:在给定可信逻辑时间和本地事务顺序下,裁决不会回拨。我们没有测真实网络延迟,没有测 NTP 故障,也没有实现一个保证按时醒来的后台扫描服务。没有工作进程来调用 expire 或提交信号时,等待记录可以一直停在 waiting;存下 deadline 不等于存在活跃调度器。
唤醒不能只靠一次内存通知#
信号事务提交后,如果代码只在内存里发一个通知,进程可能恰好在通知前退出。数据库已经说“信号有效”,但那个下一步从来没有被触发。这是等待系统自己的回执与派发缺口。
因此本例把 continuation 当作持久的“可以继续”记录,与等待终态一起提交。它不靠原协程还活着,也不要求当时有某个 worker 正在线。进程重启后,只需重新读取这些未消费记录,便能知道还有工作可做。
resume 演示最后一段本地闭环:在同一事务里写一条合成 effect,再把 continuation 的 consumed 置为 1。提交前进程退出,两项都回滚;提交后响应丢失,重试返回 replay,效果不再写第二次。八个进程争抢同一继续指令,只有一个得到 resumed,其余读到 replay。
这里必须守住“本地”二字。effect 是同库中的一条记录,不是远端付款或发信。如果改成“先调用外部工具,再更新 consumed”,这个项目的保证立刻不够。届时需要组合第一篇的幂等收件或第三篇的 unknown 对账,而不是因为前半段用了事务,就宣称整个 Agent exactly-once。
超时的意义也仅限这个等待:不再采用后来到达的回信,并生成一个本地超时后续。它没有撤销人的决定,没有取消对方服务器上已启动的操作,更不会自动补偿一个已经发生的副作用。继续与停止都必须有清楚的作用范围。
用真实退出和争用检查完整生命周期#
项目实际通过 26 项 unittest。它们检查登记前置条件、同 ID 不同内容、跨代冲突、早到拒收的永久重放、严格截止边界、时钟高水位、旧定时请求、重开前的继续消费,以及 UTF-8 和整数输入边界。
run_experiment.py 有 11 组冻结场景。它先保存等待,让另一个真实子进程接收信号,再让新进程消费继续记录;最后检查只有一条效果。这个结果证明本例不依赖原等待函数留在内存,不是生产系统重启可用性的广泛证明。
退出实验在 15 个位置调用 os._exit(73),由父进程核对实际退出码。位置覆盖等待插入、终态更新、继续指令插入、信号回执插入、本地效果插入,以及各自提交前后。提交前退出时,多表变化一起回滚;提交后退出时,回执、终态与继续记录共同保留。每次恢复后都会检查跨表不变量,再重试原操作。
并发部分不是顺序循环假装并发。每组先启动八个进程,等待全部打印 READY,再统一发 GO。分别竞争同一登记、同一等待上的多个信号、截止点的信号与超时,以及同一继续记录。结果检查成功登记一个、有效信号一个、唯一超时分支和唯一消费;不冻结谁先获得锁,也不把耗时当作吞吐指标。
在截止点那组中,信号本身已经不满足严格小于 deadline 的条件,所以无论哪个进程先拿到事务,最终都必须是 timed_out。它检验的是时间边界与原子性,不是“任何时间的信号与 timer 竞争,超时必胜”。其他时序需要按实际提交和时钟语义解释,不能从这一组实验扩张结论。
另附 audit_invariants.py 作为独立检查入口,实际运行 50 个种子、每个 500 次混合操作,共 25,000 次;再运行 40 轮、每轮 12 个 READY 门控进程的争用。它逐步复核跨表一致性、回执与终态不变、以及每代最多一个本地效果。种子固定操作选择,不固定操作系统调度,哪种终态赢了多少次不作为胜率或冻结指标。它与单元测试分开执行;这些都是合成域内的协议检查,不是模糊测试覆盖率证明、生产压力测试或恶意管理员防护。
下载运行,再决定怎样接入真实系统#
将以下六个文件保存到同一目录,用 Python 3.10 或更新版本运行。没有第三方 Python 依赖,不需要模型或 API 密钥;数据库都创建在临时目录中。
- signals.py:等待、信号、超时与本地继续协议
- test_signals.py:26 项输入与协议测试
- run_experiment.py:11 组场景及真实进程退出
- audit_invariants.py:独立混合操作与并发审计
- results.json:实际执行后保存的场景结果
- README.md:复现说明与适用边界
python3 test_signals.py
python3 run_experiment.py --check
python3 audit_invariants.py
前两个命令应分别输出 Ran 26 tests、OK 和 PASS: 11 scenario groups match results.json;独立审计输出包含 status: pass 的 JSON。实验的 --check 会重新运行子进程与数据库操作,再比较不变量;它不是读取旧 JSON 后直接输出通过。不带该参数会有意更新冻结文件,只适合重新建立经过审查的基线,不能用来消除不理解的失败。
理解代码可以从这段顺序流程开始:
from pathlib import Path
from tempfile import TemporaryDirectory
import signals as s
with TemporaryDirectory() as folder:
db = Path(folder) / 'wait.db'
s.init(db)
assert s.open_wait(db, 'case-a', 'details', 0, 10, 0)['generation'] == 1
assert s.signal(db, 'event-a', 'case-a', 'details', 1, 'ready', 3)['result'] == 'accepted'
assert s.resume(db, 'case-a', 'details', 1) == 'resumed'
assert s.open_wait(db, 'case-a', 'details', 1, 20, 4)['generation'] == 2
assert s.signal(db, 'old-event', 'case-a', 'details', 1, 'ready', 5)['result'] == 'stale_generation'
assert s.check_invariants(db)
Temporal 的 Python 消息文档区分 Signal 与 Update:Signal 是异步消息,客户端响应不等待工作流处理完成;Update 可以等待接受或拒绝,并进一步等待处理结果。这帮助我们看到“入口响应”与“业务完成”本来就需要不同语义。本文没有运行 Temporal,也没有实现它的事件历史、持久定时器或工作流重放,不能把本例的 accepted 直接映射成其 SDK 的某个完整保证。
本项目永久保留等待与回执,没有清理策略、生产容量控制、多租户鉴权、跨节点复制和通用调度器。真实接入前,要先回答消息从哪里来、谁确定地址与时间、未消费继续记录由谁扫描、断连后如何对账,再决定需要哪一种成熟工作流服务。不能因为一次正常演示收到了“yes”,就把身份、授权和可靠性全部打包成一个绿色勾。
让 Agent 等回信,关键不是把程序暂停得足够久,而是让它在醒来之后知道:这封信答的是哪一轮问题,自己是否还应当采用它,以及下一步究竟有没有已经做过。
本文由 AI 辅助制作。代码、冻结结果与进程实验经过实际运行;回信、流程和时间为合成样例,未验证真实用户身份、调用模型或执行远端业务。配图按明确的白底技术信息图文字要求原创制作,不声称读取不可见的图像参考。官方资料核对日期:2026 年 10 月 24 日。

