13 KiB

Raw Blame History

title

date

Async Generator Deadlock: yield Must Precede await When await Depends on the Yielded Event

Problem

ReActEngine 的自主暂停机制（_check_autonomy_pause）存在 P0 级死锁：当 agent 在 ReAct 循环中触发超时/失败条件需要请求用户接管时，autonomy_paused 事件永远无法送达前端，导致 resume_handler 永久阻塞，agent 主循环卡死。

根本原因是 async generator 中 yield 与 await 的顺序错误——当 await 依赖被 yield 的事件已被消费时，yield 必须先于 await 执行。

Symptoms

在 IQ-Boost 自主模式触发暂停条件后，前端永远收不到 autonomy_paused 事件
WebSocket 客户端无任何错误，但 agent 主循环卡死，不再产生新 step
resume_handler 调用永远不返回（前端永远收不到事件，因此也永远发不出 resume 消息）
服务端无异常日志，表现为「静默挂起」——典型的死锁症状，最难定位
重启服务后才能恢复，但下一次触发暂停条件时再次复现

What Didn't Work

在定位死锁根因前，尝试过若干方向，均未奏效：

怀疑 resume_handler Future 未被正确 resolve：检查 WebSocket loop 的 resume 消息处理逻辑，发现 resolve 路径正确——但问题是前端从未发送 resume 消息，所以根本走不到 resolve。这一方向浪费了时间，因为它假设事件已送达。
怀疑 pending_autonomy_resumes 注册时机问题：尝试在 await resume_handler 之前提前注册 Future，但死锁依旧——因为问题不在注册时机，而在事件是否送达消费端。
怀疑 autonomy_paused 事件被 list 缓冲导致延迟：原实现将事件先 append 到 list，再在 await 之后从 list 中 yield。一度尝试在 await 之前先 yield list 中的事件，但由于 for _pev in pause_events 循环位于 await 之后，这种「提前 yield」并未真正改变事件送达顺序。这一尝试离根因最近，但因为仍在原方法内调整顺序，无法干净地表达「先发事件、再阻塞等待」的语义。
通过加日志排查：在 await resume_handler 前后加日志，确认 await 之前日志打出，之后日志从未打出——确认是 await 处永久阻塞。但日志无法解释「为什么前端不发 resume」，直到审视事件流方向才发现事件根本没送达。

关键转折点：意识到 resume_handler 是一个反向信道——它等待的事件，正是由本 generator yield 出去的事件触发的。这构成了循环依赖，而打破循环依赖的唯一方式是保证 yield 在 await 之前完成。

Solution

将原本耦合在一个 async generator 中的「检测」与「等待恢复」拆分为两个职责单一的方法，并在调用方显式控制 yield 与 await 的顺序。

原实现（死锁）

# src/agentkit/core/react.py — 原方法（已删除）
async def _check_autonomy_pause(self, step, progress, resume_handler):
    """检测自主暂停条件，yield 暂停事件并等待恢复。"""
    pause_events = []
    # ... 检测 timeout/failure 条件 ...
    if should_pause:
        pause_events.append(
            ReActEvent(event_type="autonomy_paused", step=step, data=event_data)
        )
        # 致命错误：先 await，后 yield
        should_continue = await resume_handler(resume_token, reason)
    for _pev in pause_events:
        yield _pev  # 永远走不到这里——await 已死锁
    return should_continue, pause_events

调用方：

# 调用方（死锁）— yield 发生在 handler 返回之后，为时已晚
should_continue, pause_events = await self._check_autonomy_pause(
    step, _progress, resume_handler
)
for _pev in pause_events:
    yield _pev  # 事件在 await 之后才送达消费端——死锁
if not should_continue:
    break

修复后（拆分 + 显式顺序）

# src/agentkit/core/react.py — 拆分后的纯检测方法（非阻塞、无 yield、无 await）
def _detect_autonomy_pause(self, step, progress) -> tuple[str, str, dict] | None:
    """纯检测：返回 (reason, resume_token, event_data) 或 None。
    不 await、不 yield——保持纯函数语义，便于在调用方控制顺序。"""
    # ... 检测 timeout/failure 条件 ...
    if not should_pause:
        return None
    return (reason, resume_token, event_data)


# src/agentkit/core/react.py — 拆分后的纯阻塞方法
async def _await_autonomy_resume(
    self, resume_token: str, reason: str, resume_handler
) -> bool:
    """阻塞等待用户恢复。返回 True 表示已恢复，False 表示已取消。
    由调用方保证在 yield 事件之后才调用本方法。"""
    return await resume_handler(resume_token, reason)

调用方（关键修复点——yield 必须在 await 之前）：

# src/agentkit/core/react.py — 主 ReAct 循环（修复后）
pause_info = self._detect_autonomy_pause(step, _progress)
if pause_info is not None:
    reason, resume_token, event_data = pause_info
    # 1. 先 yield：事件送达消费端（chat.py 的 async for 循环）
    yield ReActEvent(
        event_type="autonomy_paused", step=step, data=event_data
    )
    # 2. 后 await：此时消费端已转发事件给 WebSocket 客户端，
    #    前端收到 autonomy_paused 后才会发送 resume，Future 才会被 resolve
    should_continue = await self._await_autonomy_resume(
        resume_token, reason, resume_handler
    )
    if not should_continue:
        break

Why This Works

修复的核心是恢复正确的因果链。原实现的因果链是断开的：

原（断开）：await resume_handler → （永远阻塞，因为）→ yield 事件 → 前端 resume → resolve Future
                                              ↑ 这一步永远到不了

修复后的因果链是闭合的：

修复（闭合）：yield 事件 → 消费端转发 → 前端收到 autonomy_paused
           → 前端发送 resume → WebSocket loop resolve Future
           → _await_autonomy_resume 返回 → 主循环继续

技术上，async generator 的 yield 是一个协作点：它将控制权交还给消费方（本例中是 chat.py 的 async for 循环），消费方转发事件给 WebSocket 客户端，客户端再把事件推到浏览器。只有当这条链路走完，前端的 resume 消息才有可能发回。原实现将 yield 放在 await 之后，等于要求「先收到 resume，再发送暂停事件」——这是因果倒置。

拆分为 _detect_autonomy_pause（纯函数）和 _await_autonomy_resume（纯阻塞）的好处不止修复 bug：

顺序显式化：调用方代码本身就说明了「先 yield，后 await」的契约，任何后续修改都难以意外打乱顺序。
可测试性：_detect_autonomy_pause 是纯函数，可以直接断言输入输出，无需 mock 异步环境；_await_autonomy_resume 只验证恢复/取消两条路径。
关注点分离：检测逻辑（何时暂停）与等待逻辑（如何恢复）独立演化，例如未来支持「超时自动恢复」只需改 _await_autonomy_resume。

Prevention

1. 通用规则：反向信道场景下 yield 必须先于 await

当 async generator 中存在「await 等待的事件由本 generator yield 出去触发」的反向信道时，yield 必须先于 await。这扩展了项目已有的异步生成器安全规则。

在 AGENTS.md 已有规则的基础上补充：

异步生成器安全（扩展）：在 async def 中禁止在第一个 yield 之前使用 return；并且当 await 依赖被 yield 的事件已被消费时，yield 必须先于 await。

2. 拆分检测与阻塞，让顺序契约显式

凡是「先发事件、后等响应」的 async generator，优先考虑拆分为：

一个纯检测方法（同步、无 await、无 yield），返回决策结果
一个纯阻塞方法（async，只 await handler），由调用方在 yield 之后调用

这种拆分让「先 yield 后 await」从隐式约定变成显式代码结构，code review 时一目了然。

3. Code review 检查项

审查任何包含 yield 与 await 的 async generator 时，问自己一个问题：

这个 generator 里的某个 await，是否依赖于消费端已经收到了此前 yield 出去的事件？

如果答案是「是」，则必须保证 yield 在该 await 之前执行。常见的反向信道场景：

用户确认/接管（本例）
资源就绪信号（yield 请求 → await 就绪）
跨 agent handoff（yield handoff 事件 → await 对端 ack）

4. 测试：mock 消费端验证事件顺序

为「yield 后 await」契约写一个最小可执行检查（遵循 ponytail 规则——最小可失败的断言，无框架）：

# tests/unit/test_react_autonomy_pause.py — 最小自检
import asyncio
import pytest

async def _consume_then_resume(agent, steps, resume_handler):
    """模拟 chat.py 消费端：先收到 autonomy_paused 事件，才 resolve resume。"""
    received_pause = asyncio.Event()
    resume_future = asyncio.get_event_loop().create_future()

    async def fake_handler(token, reason):
        # 必须等消费端确认收到事件后才 resolve
        await received_pause.wait()
        return True

    events = []
    async for ev in agent.run_stream(steps):
        events.append(ev)
        if ev.event_type == "autonomy_paused":
            received_pause.set()  # 模拟前端收到事件后发 resume
            resume_future.set_result(True)

    # 断言：autonomy_paused 必须出现在事件流中，且 run_stream 必须能正常结束
    # （而非死锁挂起——pytest 的 asyncio 超时会捕获死锁）
    assert any(ev.event_type == "autonomy_paused" for ev in events)

这个检查在原实现下会因为 await asyncio 超时而失败（死锁），在修复后通过——正是「最小可失败的检查」。

5. 项目规则联动

本学习扩展了 .trae/rules/project_rules.md 中已记录的异步生成器安全规则。原规则只覆盖「第一个 yield 之前不能 return」这一类问题，本学习补充了「yield 与 await 的顺序依赖」这一类。两条规则应一起记忆：

第一类（已有）：return 在第一个 yield 之前 → 函数被识别为协程而非 async generator
第二类（本学习）：await 在 yield 之前，且 await 依赖被 yield 的事件 → 死锁

PR #27（feat/agent-iq-boost）：本 bug 的引入与修复均在该分支完成
AGENTS.md > 异步生成器安全：项目级规则，本学习为其扩展项
.trae/rules/project_rules.md > Python Async Generator Safety：第一类规则的完整说明与 return; yield 模式
src/agentkit/core/react.py：受影响文件，_detect_autonomy_pause 与 _await_autonomy_resume 的当前实现位置
src/agentkit/server/routes/chat.py：消费端，async for 循环转发 autonomy_paused 事件至 WebSocket 客户端
根因类别：async_timing——async generator 中 yield 与 await 的顺序在存在依赖时具有决定性

相邻学习（同一 execute_stream 合约）

以下文档覆盖 ReActEngine.execute_stream() async generator 的不同不变量，与本学习互补：

streaming-event-contract-residuals.md — CancellationToken 注册对称性（execute_stream 绕过 BaseAgent.execute() 的 _active_tokens dict，破坏协作式取消）
streaming-event-whitelist-and-accumulation.md — token + final_answer 双重累积、WS 事件白名单、async generator 中的 except-Exception 作用域
long-horizon-reliability-code-review-fixes.md — execute() 入口未调用 reset() 导致状态隔离失败

13 KiB Raw Blame History Unescape Escape