# 持久信号与超时竞争：等回信的 Agent，重启后还记得在等谁吗？

> 把外部回信、等待代次、截止时间和后续执行放进可检查的协议：用 SQLite、真实进程退出与并发竞争，验证早到拒收、重复回执、迟到隔离与唯一继续。

- 作者：芝士AI吃鱼
- 发布日期：2026-10-24
- 主题：Agent、AI工程、工作流、事件驱动、SQLite
- HTML 正文：[https://ai-knowledgepoints.cn/blog/agent-reliability-16-durable-signals-timeouts](https://ai-knowledgepoints.cn/blog/agent-reliability-16-durable-signals-timeouts)
- Markdown 永久链接：[https://ai-knowledgepoints.cn/blog/agent-reliability-16-durable-signals-timeouts/index.html.md](https://ai-knowledgepoints.cn/blog/agent-reliability-16-durable-signals-timeouts/index.html.md)

引用本文时，请注明作者与 HTML 正文链接。Markdown 版本用于机器阅读，与 HTML 正文共享同一内容来源。

## 正文

Agent 把问题发出去，等人回一句话。人去开会了，服务却在这时候重启。几小时后，回复终于到了：界面告诉对方“已收到”，后台那个正在等待的协程，早已不在了。

这是本文构造的故障场景，不是真实用户事故。它提醒我们：让一个函数暂停，不等于把“等谁、等哪一轮、等到什么时候”保存下来。还有一种更难察觉的错误：旧问题的回信晚到了，恰好唤醒了新问题。消息没有丢，流程却接错了话。

[第十五篇](https://ai-knowledgepoints.cn/blog/agent-reliability-15-version-pinned-workflows)固定的是任务执行的定义，本篇作为 Agent 可靠执行实战第十六篇，处理长任务在外部消息与截止时间之间的完整等待生命周期：登记、拒收早到消息、接受有效信号、裁决超时、保存继续指令、重启恢复和再次等待。

随附项目不调用大模型，不发送邮件，不连接真实审批系统。回信是合成字符串，时间是可信的逻辑整数；真正执行的是 SQLite 事务、操作系统子进程退出与并发争用。它不把“能收到一个信号”当作真人身份验证，也不替代[第八篇的精确意图批准](https://ai-knowledgepoints.cn/blog/agent-reliability-08-intent-bound-approvals)。批准是否合法与回信是否交给了正确等待，是两层不同的问题。

## 等待要有地址，而且地址不能只写流程号

示例给每次等待三个标识：`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 的事务文档](https://www.sqlite.org/lang_transaction.html)说明，同一数据库可以有多个读事务，但同时只有一个写事务；IMMEDIATE 会立即尝试开始写事务，并可能遇到 SQLITE_BUSY。本文利用的是这个单库序列化边界，不是“分布式事务天然存在”。连接的等待超时只是便于复现小规模争用，锁错误依然会暴露，不能解释成业务超时已完成。

信号进来后，若当前代已换，记 stale_generation；若该代已终结，记 closed；若还是 waiting，却已满足 `now >= deadline`，就不能采用回信。此时信号处理本身会将等待终结为 timed_out，并生成超时继续指令，自己的回执为 too_late。

最后这条很重要：不能让“超时扫描器还没跑”变成“截止时间仍然有效”。定时任务晚执行，不应该给某条回信随机延长契约。恰好在 deadline 的信号也不接受，本例采用严格小于截止时间的规则。这个边界需要写进协议，不能依赖某台机器碰巧先调度谁。

如果有效信号先提交，后来的 expire 看到 closed，不再写另一个超时分支。如果超时先终结，旧时间戳的信号也不能把终态翻回来。状态、继续指令及本地效果的唯一键共同让不一致显露为错误，而不是事后从两个已运行的分支里挑一个“正确结果”。

![持久等待由流程、步骤和代次定位；有效信号与到期检查在同一数据库事务内竞争，唯一终态生成唯一继续指令，迟到或旧代信号不唤醒新一轮](https://ai-knowledgepoints.cn/images/articles/agent-reliability-16/durable-wait.svg)

原创协议示意。时间为可信逻辑值，信号不是身份认证；继续指令只覆盖本地效果。

[查看竖版大图](https://ai-knowledgepoints.cn/images/articles/agent-reliability-16/durable-wait-mobile.svg)

## 时钟也有状态，不能把旧时间拿来洗白

为了让测试不依赖真实等待，示例的 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”，这个项目的保证立刻不够。届时需要组合[第一篇的幂等收件](https://ai-knowledgepoints.cn/blog/agent-reliability-01-outbox-idempotency)或[第三篇的 unknown 对账](https://ai-knowledgepoints.cn/blog/agent-reliability-03-unknown-reconciliation)，而不是因为前半段用了事务，就宣称整个 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：等待、信号、超时与本地继续协议](https://ai-knowledgepoints.cn/examples/agent-reliability-16/signals.py)
- [test_signals.py：26 项输入与协议测试](https://ai-knowledgepoints.cn/examples/agent-reliability-16/test_signals.py)
- [run_experiment.py：11 组场景及真实进程退出](https://ai-knowledgepoints.cn/examples/agent-reliability-16/run_experiment.py)
- [audit_invariants.py：独立混合操作与并发审计](https://ai-knowledgepoints.cn/examples/agent-reliability-16/audit_invariants.py)
- [results.json：实际执行后保存的场景结果](https://ai-knowledgepoints.cn/examples/agent-reliability-16/results.json)
- [README.md：复现说明与适用边界](https://ai-knowledgepoints.cn/examples/agent-reliability-16/README.md)

```bash
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 后直接输出通过。不带该参数会有意更新冻结文件，只适合重新建立经过审查的基线，不能用来消除不理解的失败。

理解代码可以从这段顺序流程开始：

```python
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 消息文档](https://docs.temporal.io/develop/python/workflows/message-passing)区分 Signal 与 Update：Signal 是异步消息，客户端响应不等待工作流处理完成；Update 可以等待接受或拒绝，并进一步等待处理结果。这帮助我们看到“入口响应”与“业务完成”本来就需要不同语义。本文没有运行 Temporal，也没有实现它的事件历史、持久定时器或工作流重放，不能把本例的 accepted 直接映射成其 SDK 的某个完整保证。

本项目永久保留等待与回执，没有清理策略、生产容量控制、多租户鉴权、跨节点复制和通用调度器。真实接入前，要先回答消息从哪里来、谁确定地址与时间、未消费继续记录由谁扫描、断连后如何对账，再决定需要哪一种成熟工作流服务。不能因为一次正常演示收到了“yes”，就把身份、授权和可靠性全部打包成一个绿色勾。

让 Agent 等回信，关键不是把程序暂停得足够久，而是让它在醒来之后知道：这封信答的是哪一轮问题，自己是否还应当采用它，以及下一步究竟有没有已经做过。

系列讨论可查看[知识星球](https://ai-knowledgepoints.cn/planet)，或通过[公众号入口](https://ai-knowledgepoints.cn/about#wechat)关注更新。

本文由 AI 辅助制作。代码、冻结结果与进程实验经过实际运行；回信、流程和时间为合成样例，未验证真实用户身份、调用模型或执行远端业务。配图按明确的白底技术信息图文字要求原创制作，不声称读取不可见的图像参考。官方资料核对日期：2026 年 10 月 24 日。