DeerFlow 阻塞式 IO 防护：如何编写通过"牙齿验证"的运行时 Anchor 测试

本文基于 DeerFlow 仓库中的 .agent/skills/blocking-io-guard/references/good-anchor-rules.md 展开，讲清楚在 asyncio 事件循环上防止同步阻塞 IO 的最后一道防线——运行时 anchor（锚点测试）——的编写标准、"牙齿验证"（teeth check）验收方法，以及何时才允许新增 Blockbuster 项目级规则。读完后，你将能够为 DeerFlow 后端的任意异步阻塞 IO 隐患写出可被 CI 严格门禁（make test-blocking-io）拦截回归的 anchor 测试，并掌握 anchor 与 runtime rule 两条路线的判别准则。

1. Anchor 在整个阻塞 IO 防护体系中的位置

DeerFlow 的阻塞 IO 防护由静态与运行时两套检测器协同完成，二者职责不同（见 backend/docs/BLOCKING_IO_DETECTION.md）：

• 静态检测器是发现工具：make detect-blocking-io 扫描后端源码，把可疑的阻塞调用点写入 .deer-flow/blocking-io-findings.json，静态 finding 是"候选"，不是生产环境真的阻塞了事件循环的证明；
• 运行时检测器是 CI 回归门禁：基于 Blockbuster，当 app.* 或 deerflow.* 作用域下的代码在 asyncio 事件循环线程上执行阻塞 IO 时让测试失败。

关键约束在于：运行时检测只能看见被测试实际执行到的路径。如果某条生产阻塞路径没有任何测试去驱动它，门禁再严格也形同虚设。Anchor（锚点测试）就是填补这个缺口的东西——它把生产环境的真实异步入口点在门禁下"压"一遍，让 CI 从此守住这条路径。

anchor 统一放在 backend/tests/blocking_io/ 目录，目录中的 conftest.py 用 hookwrapper 把整个 pytest item 生命周期（setup + call + teardown）包进严格的 detect_blocking_io_strict() 门禁——这意味着连 async fixture 和 lifespan 代码里的阻塞 IO 也会被抓到，而不只是测试函数体内。conftest 还通过路径过滤确保严格门禁只对 backend/tests/blocking_io/ 下的用例生效，并提供 @pytest.mark.allow_blocking_io 标记作为显式豁免。

门禁的作用域定义在 backend/tests/support/detectors/blocking_io_runtime.py：

_SCANNED_MODULES: tuple[str, ...] = ("app", "deerflow")

_PROJECT_BLOCKING_RULES: tuple[tuple[str, BlockBusterFunction], ...] = ()

@contextmanager
def detect_blocking_io_strict() -> Iterator[BlockBuster]:
    """Activate Blockbuster scoped to app.* and deerflow.* callers only."""
    bb = BlockBuster(scanned_modules=list(_SCANNED_MODULES))
    _install_project_rules(bb)
    try:
        bb.activate()
        yield bb
    finally:
        bb.deactivate()

从源码结构看，scanned_modules=("app", "deerflow") 的设计让 pytest、langchain、importlib 等第三方库的调用不在扫描范围内，只有调用栈穿过 app.* 或 deerflow.* 业务代码的阻塞 IO 才会抛出 BlockingError——这避免了测试基础设施本身产生大量误报。这也解释了为什么 anchor 模板里可以大放心使用 tmp_path 写测试侧的 IO：测试模块本身不在 scanned_modules 内，门禁看不见它。

2. 好的 anchor 的五条判定标准

good-anchor-rules.md 给出的核心定义如下（原文五条，逐条展开）：

1. 调用真实的"生产异步入口点"，而不是底层 helper——除非那个 helper 本身就是生产执行的入口点。
2. 不得用测试专属的 asyncio.to_thread / run_in_executor 包装绕过被守护的阻塞面。也就是说，不能在测试里"替生产代码 offload"，否则门禁永远看不到阻塞。
3. 当缺陷形态是文件系统 IO 时，使用真实的本地文件系统作为输入（如 pytest 的 tmp_path）。
4. 只 mock 外部依赖边界（网络服务、第三方 saver 等），永远不要 mock 被守护的 offload 机制本身。
5. 驱动你要保护的具体分支（error / cleanup / 404 / 409），而不仅仅是 happy path。

文档还特别提醒：该文件与 templates/anchor.template.py 中的示例全部是"文件系统风味"的，它们演示的是怎么写测试，而不是 SOP 只覆盖文件系统场景——同样的规则适用于检测器报告的每一类问题（FILE_IO、HTTP、SUBPROCESS、SLEEP），验收标准永远是下文的"牙齿检查"，绝不是与示例的相似度。

第 4 条尤其值得展开：如果 mock 掉了 offload 本身（例如 mock 掉 asyncio.to_thread），anchor 就从"守护生产路径"退化成了"守护调用次数"，未来重构 offload 方式（比如从 to_thread 换成 run_in_executor）会导致假失败，而真正回到循环上的阻塞反而漏过。

3. 牙齿验证（Teeth Check）：anchor 的验收测试

这是整篇规则文档的灵魂。原文的标准表述是："一个没有证明过 RED 的 green-on-happy-path anchor 是假覆盖。不要合入它。"

验收流程三步：

1. 重新引入阻塞（GUARD 场景：临时回滚 offload；FIX+ANCHOR 场景：在修复前的代码上运行）；
2. 执行 cd backend && make test-blocking-io → anchor 必须失败（RED）；
3. 恢复修复 → anchor 必须通过（GREEN）。

Makefile 中该目标的实际定义（backend/Makefile）是：

test-blocking-io:
	PYTHONPATH=. PYTHONIOENCODING=utf-8 PYTHONUTF8=1 uv run pytest tests/blocking_io -q --tb=short

即整体验收就是跑 backend/tests/blocking_io/ 全目录，也可按测试名定向到单个 anchor。"牙齿"（teeth）这个词的语义是：门禁必须真的能咬人——如果故意把代码改回阻塞版本后门禁纹丝不动，说明这条 anchor 保护了个寂寞，或说明 Blockbuster 没有对应规则（后者引出第 5 节的 RULE 路线）。

3.1 一个符合标准的真实 anchor 示例

backend/tests/blocking_io/test_jsonl_run_event_store.py 是上述五条标准的完整落地（对应修复 issue #3084）：

"""Regression anchor: JsonlRunEventStore async API must not block the loop. ..."""
import pytest

pytestmark = pytest.mark.asyncio


async def test_jsonl_run_event_store_async_api_does_not_block_event_loop(tmp_path: Path) -> None:
    from deerflow.runtime.events.store.jsonl import JsonlRunEventStore

    store = JsonlRunEventStore(base_dir=str(tmp_path))

    # 测试侧用真实文件系统准备种子数据（tmp_path），
    # 测试模块不在 scanned_modules 内，这些 IO 对门禁不可见。
    thread_dir = tmp_path / "threads" / "t1" / "runs"
    thread_dir.mkdir(parents=True, exist_ok=True)
    (thread_dir / "r0.jsonl").write_text(
        '{"seq": 1, "category": "message", "run_id": "r0"}\n', encoding="utf-8"
    )

    # 驱动真实生产异步入口点：put / put_batch / put_if_absent /
    # list_* / count / delete_* —— 任何一条在循环上重新阻塞都会让门禁失败
    record = await store.put(thread_id="t1", run_id="r1",
                             event_type="message", category="message", content="hi")
    assert record["seq"] >= 2
    ...

对照标准逐条验证：调用的是生产真实实现的 JsonlRunEventStore 异步 API（标准 1）；没有测试侧 to_thread 包装（标准 2）；用 tmp_path 真实文件系统输入（标准 3）；没有任何外部依赖需要 mock，更没有 mock offload（标准 4）；覆盖了写、读、删、幂等插入（put_if_absent 二次插入的 created is False 分支）、以及 list_events 带/不带 event_types 过滤两个分支——不只有 happy path（标准 5）。其 docstring 明确说明意图：守护的不是"某一次 to_thread 调用的存在"，而是"这些方法中任何一条在事件循环上重新引入阻塞 IO 都会让 CI 失败"。

仓库中还有大量同模式的 anchor 可直接参考：SQLite checkpointer 初始化（test_sqlite_lifespan.py）、uploads 目录扫描（test_uploads_middleware.py）、各渠道文件接收（test_dingtalk_receive_file.py、test_feishu_receive_file.py）以及门禁自身的健康检查（test_gate_smoke.py，验证未 offload 调用会被抓到、opt-out 标记有效、异常后补丁能恢复）。

3.2 模板：从 anchor.template.py 起步

新建 anchor 时，从 .agent/skills/blocking-io-guard/templates/anchor.template.py 复制改造即可。模板骨架：

"""Template: a tests/blocking_io/ runtime anchor.
Copy into backend/tests/blocking_io/test_<area>.py and adapt. The suite's
conftest already wraps every test here in the strict Blockbuster gate, so you do
NOT import or activate the detector — just drive the real async entry point.
"""
import pytest

# from app.<module> import <real_async_entry_point>

pytestmark = pytest.mark.asyncio


async def test_<entry_point>_offloads_blocking_io_on_<branch>(tmp_path: Path) -> None:
    # Arrange: real inputs at the boundary the code blocks on (FS -> tmp_path;
    #   HTTP/subprocess -> stub the external service). Mock ONLY the external
    #   boundary, never the offload under test.
    # Act + Assert: call the REAL production async entry point and drive the
    # specific branch you are guarding (e.g. force a failure to hit the cleanup
    # path). If the entry point performs blocking IO on the loop, the gate fails.
    #   await <real_async_entry_point>(...)
    raise NotImplementedError("Replace with the real async entry point call.")

模板注释里有两个关键约定：conftest 已接管门禁激活，不要在 anchor 里 import 或激活检测器；测试名用 test_<entry_point>_offloads_blocking_io_on_<branch> 的命名模式，把守护对象和分支写进名字。

4. RULE 路线：什么时候才允许新增项目级规则

Blockbuster 内置规则已能覆盖常见的阻塞原语。规则文档明确指出，SOP 中存在两个有意的开口：

1. 覆盖开口（常规情况）：规则已经看得见这个原语——你只需要一个 anchor，让运行时检测真正执行到业务路径，并让 CI 防止回归。
2. 规则开口（罕见）：你重新引入了真实的阻塞，而门禁仍然 GREEN——Blockbuster 对该原语没有规则。

只有第 2 种情况才走 RULE 路线。项目规则放在 backend/tests/support/detectors/blocking_io_runtime.py 的 _PROJECT_BLOCKING_RULES 中（当前仓库里该元组为空，即尚未有项目级规则）。示例形态：

import subprocess

from blockbuster import BlockBusterFunction

_PROJECT_BLOCKING_RULES = (
    (
        "subprocess.Popen.__init__",
        BlockBusterFunction(
            subprocess.Popen,
            "__init__",
            scanned_modules=["app", "deerflow"],
        ),
    ),
)

4.1 准入标准（admission criteria）

新增项目级规则的影响面是整个 blocking-IO 套件（global blast radius），因此准入条件被写得很硬：

• 必须持有"fails-to-fail"的 anchor 作为证据：一个符合上文标准的好 anchor，驱动了一条真实阻塞的路径，却仍然 GREEN。没有证据，就不许加规则（No evidence, no rule）。
• 该原语必须是真实的阻塞调用——对照其实现或文档核实过，而不是静态检测器的误报。
• 规则必须单独提交（own commit），提交信息中写明：该原语、暴露缺陷的 anchor、以及对整个套件的影响。新增规则后要跑完整的 make test-blocking-io 套件——新规则可能把其他原本 GREEN 的测试打红，而每一个这样的 RED 二选一：要么是真实存在的潜在 bug（修它），要么是规则过度（收窄规则）。
• 如果你无法承担这个爆炸半径（例如外部贡献者），带着证据升级到维护者处理，而不是自己加规则。

文档最后一条是整份规则的压舱石：

绝不因为某条路径没有被测试就新增运行时规则——那种情况需要的是 anchor，不是规则。

从 sop-skeleton.md 的抽象视角看，这条纪律对应 SOP 第 5 步"验证牙齿"的判别逻辑：一个"真正有问题却保持绿色"的模式，是 rule 信号，而不是覆盖成功。该骨架还把整个流程泛化为六个域无关步骤（Scope 划定范围 → Judge 路由 → Fix + re-scope → Generate → Verify teeth → Deliver），供未来第二个检测器域复用；当前仓库只有 blocking-IO 一个域实例。

5. 实操速查：anchor 与 rule 的判别流程

把规则文档浓缩成一张可执行的判别表：

现象	判定	动作
阻塞路径有测试执行到，门禁 RED→GREEN 复现	正常覆盖	保留 anchor，随变更提交
生产路径无测试执行到，原语本身已被规则覆盖	覆盖开口	新建/扩展 anchor（标准 1–5 + 牙齿验证）
故意重新引入阻塞，门禁仍 GREEN	规则开口	持有 fails-to-fail 证据，按准入标准单独提交项目级规则
某路径只是没被测试	非规则问题	写 anchor，禁止加规则

配套的执行入口（与 SKILL.md 中 SOP 的 Step 0 / Step 5 对应）：

# 静态候选扫描（全库 triage 模式，输出 .deer-flow/blocking-io-findings.json）
make detect-blocking-io

# 运行时门禁（牙齿验证的唯一判据）
cd backend && make test-blocking-io

总结一句：在 DeerFlow 中，anchor 的价值不由"测试写得像不像示例"决定，而由"把阻塞改回去时 CI 是否真的变红"决定；runtime rule 则是高门槛的最后手段——先有 fails-to-fail 的 anchor 证据，再谈动全套件共享的规则面。