last30days-skill：为什么 ThreadPoolExecutor 上的"墙钟预算"是幻觉——非守护工作线程会在解释器退出时被 join

这篇技术文章基于仓库中的缺陷复盘文档 非守护执行器线程会让墙钟预算失效，围绕 last30days-skill 发现模式（discovery）的 enrichment 扇出环节，讲清楚一个隐蔽但后果严重的 Python 并发问题：ThreadPoolExecutor 的工作线程是非守护（non-daemon）线程，会在解释器退出时被统一 join——这意味着 as_completed(timeout=...) 只能约束"答案何时返回"，却约束不了"进程何时退出"。读完本文，你将掌握 daemon 线程 + Semaphore + 结果队列 + 单调时钟截止期的正确写法，以及用一行断言把"线程必须是守护线程"这个隐含契约钉进测试的验收方法。

问题背景：一个约束结果却不约束进程的预算

last30days-skill 的 discovery 流程分两个阶段：阶段一从各信息源提名候选话题（nomination），阶段二 enrich_nominations() 对每个提名话题跑一次完整的多源研究子流程（sub-run），再汇总排序。阶段二会打到与正常研究流程完全相同的上游 API（Reddit、X、YouTube、HN、arXiv、Polymarket 等），因此设计上必须同时满足两点：

1. 并行度保持很低（默认 ENRICH_MAX_WORKERS = 3），避免对上游 API 造成压力；
2. 整个批次受一个墙钟预算（ENRICH_BUDGET_SECONDS，默认 240 秒）约束——超时的话题直接降级为"仅提名"（nomination-only），绝不致命。

这两个常量定义在 pipeline.py：

# Enrichment fan-out bounds. Sub-runs hit the same upstream APIs as a normal
# research pass, so parallelism stays low and the whole batch runs against a
# wall-clock budget - a slow topic is dropped, never fatal.
ENRICH_LIMIT = 6
ENRICH_DEPTH = "quick"
ENRICH_MAX_WORKERS = 3
ENRICH_BUDGET_SECONDS = 240.0

第一版实现把预算建在 ThreadPoolExecutor 上——它看起来像预算执行，实际上预算只约束了"答案"，没有约束"进程"。只要有一个子流程卡死（例如一次不触发超时、一直停住的联网请求），整个 CLI 就会在批次"完成"之后无限期挂起。

症状：所有单元测试都通过，进程却退不出

复盘文档记录的三个症状值得逐条理解，因为每一类都代表一个常见的误判来源：

• 批次"完成"但进程不退出：只要有任一下子流程真的挂死，CLI 进程就会越过 enrichment 预算继续存活，甚至无限期存活。批次结果照常组装完毕，但 Python 在解释器退出时会 join executor 的非守护线程，于是进程卡死。
• 只在真正挂死的 worker 下才可见：慢话题的单元测试观察到的正是 as_completed(timeout=...) 所承诺的一切——快话题返回、慢话题从结果中丢弃——而且测试进程本身退出良好，因为测试里的"慢" worker 只是 sleep 几秒，并不是真的挂死。这个缺陷完全活在"进程生命周期"这一层，而面向结果（result-oriented）的测试永远碰不到这一层。
• 由代码评审在发布前捕获：该问题在 PR #816 的评审中被标记为 P1（"Enrichment Budget Keeps Running"），修复随 v3.14.0 发布（对应 CHANGELOG.md 中 2026-07-12 的版本条目）。

这里有一个对 Agent/自动化流水线尤其危险的隐含后果：预算到期后进程不退，任何"跑完即走"的调度假设都会被打破——挂住的不只是这一次运行，而是持有进程的所有上游逻辑。

为什么三个"预算旋钮"全都救不了运行中的线程

第一版实现大致是这样：

executor = ThreadPoolExecutor(max_workers=max_workers)
futures = {executor.submit(_run_one, n): n for n in nominations}
try:
    for future in as_completed(futures, timeout=budget_seconds):
        ...collect result...
finally:
    executor.shutdown(wait=False, cancel_futures=True)

这段代码里每个旋钮都真实地做了一点事——只是做的都不是需要的那件事。复盘文档给出了逐项的失效机理：

1. as_completed(futures, timeout=budget) 只约束消费者。超时到期时它在收集循环里抛出 TimeoutError，对 worker 不发送任何信号。Python 线程无法从外部杀死，所以一个运行中的子流程会继续跑下去。
2. cancel_futures=True 只能取消还排队中的 future——即可调用对象尚未开始执行的 future。对处于 RUNNING 状态的 future，Future.cancel() 返回 False；正在执行它的 worker 毫发无损。
3. shutdown(wait=False) 只是跳过在 shutdown 调用时刻 join 线程，并不解绑（detach）它们。CPython 的 ThreadPoolExecutor 创建工作线程时是非守护的，并且（自 Python 3.9 起）通过 threading._register_atexit 钩子注册了"解释器退出时 join 所有 worker"的行为。因此即便调用了 shutdown(wait=False)，解释器退出依然会阻塞到挂死的 worker 返回——对于一个卡住的 fetch，这可能意味着永远。

综合效果就是复盘文档里那句精准的描述：预算到期了，慢话题被正确地报告为 nomination-only，而进程依然原地坐在那里，被那个预算"以为"已经抛弃的线程死死撑住。

一句话总结这个类别的错误：as_completed(timeout=...) + shutdown(wait=False, cancel_futures=True) 是答案有界（answer-bounding）的，不是进程有界（process-bounding）的。如果需求是"这个批次不得延长进程的生命"，非守护的 executor 线程（CPython 自 3.9 起的行为，executor 线程从 atexit-daemon 处理改为 threading._register_atexit join）无法兑现它，而且没有任何旋钮组合能改变这一点。

修复方案：daemon 线程 + Semaphore + 结果队列 + 单调截止期

实际修复位于 pipeline.py 的 enrich_nominations()，用四个构件替换了 executor：

# Daemon threads + a semaphore instead of ThreadPoolExecutor: executor
# threads are non-daemon and joined at interpreter shutdown, so one hung
# sub-run could keep the whole process alive long after its topic was
# downgraded to nomination-only. Daemon workers make the wall-clock budget
# real - stragglers cannot delay process exit. Abandonment is safe because
# internal_subrun passes write nothing to disk (no save, no library sync,
# no store), and every fetch layer inside run() carries its own timeout.
enriched: dict[str, EnrichedTopic] = {}
results_queue: queue.Queue[tuple[Nomination, schema.Report | None, Exception | None]] = queue.Queue()
slots = threading.Semaphore(max(1, max_workers))

def _worker(nomination: Nomination) -> None:
    with slots:
        try:
            results_queue.put((nomination, _run_one(nomination), None))
        except Exception as exc:  # noqa: BLE001 - containment is the contract
            results_queue.put((nomination, None, exc))

for nomination in nominations:
    threading.Thread(
        target=_worker,
        args=(nomination,),
        name=f"discover-enrich-{nomination.name[:32]}",
        daemon=True,
    ).start()

deadline = time.monotonic() + max(1.0, budget_seconds)
pending = len(nominations)
while pending and (remaining := deadline - time.monotonic()) > 0:
    try:
        nomination, report, exc = results_queue.get(timeout=min(remaining, 0.5))
    except queue.Empty:
        continue
    pending -= 1
    ...record EnrichedTopic success or error...
# Budget expired (or all done): unfinished topics fall through below as
# nomination-only; their daemon workers are abandoned and cannot block exit.

从源码看，各构件的职责划分非常清晰：

• daemon 线程负责"进程有界"：daemon=True 的 worker 不在解释器退出时被 join，一个 fetch 到一半的 daemon worker 直接随进程消亡。这才让墙钟预算成为现实——到期就意味着"进程现在可以退出"，而不是"等掉队者跑完之后再退出"。
• Semaphore(max(1, max_workers)) 保留 executor 唯一有用的性质：每个 worker 在 with slots: 内执行，在途子流程数被钳制在 max_workers；超出上限的线程虽然存在，但阻塞在信号量上，成本几乎为零。上游 API 看到的并行度与之前完全一致。
• time.monotonic() 截止期独立于 worker 行为约束消费者：results_queue.get(timeout=min(remaining, 0.5)) 每秒至少唤醒两次重新检查截止期，因此无论任何 worker 在做什么，收集循环都会在预算到期后约 0.5 秒内退出。
• 结果队列统一承载成功与失败：worker 内部捕获异常并以 (nomination, None, exc) 入队（"containment is the contract"），批次永不向上抛出异常。

消费循环退出后的收尾（pipeline.py L1017-L1034）按提名顺序回补结果：所有仍没有对应条目的话题都会以 error="enrichment budget exhausted" 落到 nomination-only，并向 stderr 打印降级说明；批次保持提名顺序、绝不抛出异常。

"丢弃 worker 是安全的"这个前提从哪里来

复盘文档特别强调：daemon 放弃（abandonment）之所以安全，是因为一个写在代码注释里、放在下一个编辑者必经之处的免写前提——enrichment 子流程是 internal_subrun=True 路径，它什么都不写盘（no save, no library sync, no store），并且 run() 内部的每个 fetch 层都自带超时。源码印证了这一点：

• _run_one() 调用 run(..., internal_subrun=True)（pipeline.py L956-L966）；
• internal_subrun=True 时子流程的线程池上限被压低——_inner_max_workers 中子流程最多 4 个内部 worker（顶层运行是 4~16 个），避免六路竞争者扇出把线程数推到 ~96；
• 子流程同样被排除在 library context 解析之外（internal_subrun 为真时直接跳过，见 pipeline.py L1906-L1910）。

反过来，文档也划出了明确的反例边界：会变更共享状态（文件、数据库、缓存）的 worker 绝不能用这种方式放弃，它们需要协作式取消（cooperative cancellation）。这个前提被刻意向外公开：如果将来有人在 worker 里加了一次磁盘写入，就会撞上这条注释警告。

测试如何把这个契约钉死

仅靠"修复代码"是不够的——因为原始缺陷恰恰是对结果导向测试不可见的。tests/test_discover_enrich.py 用一组测试把进程生命周期属性直接变成断言对象：

1. test_enrich_workers_are_daemon_threads（L117-L132）：mock 掉 run()，在每个 worker 内部读取 threading.current_thread().daemon 并收集。daemon 属性被直接测试，而不是从进程行为推断——这正是复盘文档"显式测试 daemon-ness"预防项的落地。
2. test_enrich_concurrency_capped_by_semaphore（L135-L156）：6 个提名、max_workers=2，用锁记录在途 worker 计数，断言峰值从不超过 2——验证 Semaphore 真的承担了并发上限职责。
3. test_enrich_budget_expiry_drops_slow_topic_to_nomination_only（L79-L95）：一个快速话题和一个 5 秒话题在 budget_seconds=1.0 下运行，断言快速话题返回完整 report，慢话题降级为 nomination-only 且错误信息包含 "budget"。
4. test_enrichment_path_never_uses_thread_pool_executor（L503-L512）：这是一个源码级回归钉——直接 inspect.getsource(pipeline.enrich_nominations)，断言函数体内不出现 ThreadPoolExecutor( 构造、且包含 daemon=True 和 Semaphore。它把"这个反模式不许回来"变成了可执行的 CI 检查，并注明了本文档路径作为出处。

值得注意的还有 tier 参数化的配套测试：resume 流程存在 shallow/deep 两级参数（one-shot 保持 quick/240/3，deep 档升级为 default/450(可配)/4，见 pipeline.py L1576-L1610 的 RESUME_DEEP_ENRICH_* 常量与 LAST30DAYS_ENRICH_BUDGET_SECONDS 配置读取）。test_discover_enrich.py 中 deep 档同样有一套 daemon/信号量/预算降级的镜像测试（如 test_enrich_workers_are_daemon_threads_at_deep_tier），保证参数化路径不会把守护线程契约丢掉。

为什么这个方案"恰好"够用：逐条机理对照

机制	它约束的对象	为什么它成立
daemon=True worker	进程生命周期	CPython 退出序列只等待非守护线程；daemon worker 随进程消亡，预算到期即可以退出
time.monotonic() 截止期	消费者	与 worker 行为解耦；get(timeout=min(remaining, 0.5)) 每秒至少重检两次，~0.5s 内必退出
Semaphore(max_workers)	上游 API 压力	在途子流程数被钳制，超出上限的线程仅阻塞在信号量上
免写前提（internal_subrun）	放弃的安全性	无 save、无 library sync、无 store；进程退出杀掉这类 worker 不可能损坏任何状态
worker 内 per-request 超时	挂死的概率	每个 fetch 层自带超时，使"挂住的 worker"成为罕见病态而非常态

预防清单：把"掉队者问题"写进设计与评审

复盘文档最后的预防章节给出了一份可以直接搬进自己项目的清单，这里完整继承其要点：

• 任何"预算/超时"覆盖线程工作时，必须明确说明一个 RUNNING 状态的掉队者会怎样。 如果设计文档或注释只说了"结果会怎样"，那么进程生命周期这个问题就是未回答的——而默认答案（非守护线程在退出时被 join）对 CLI 通常是错的。
• 对可放弃的工作优先用显式 daemon 线程。 当掉队者可以安全丢弃时，threading.Thread(daemon=True) + Semaphore + 队列 + 单调截止期，代码量与 executor 方案相差不大，而且真正执行了预算。ThreadPoolExecutor 应保留给"你打算等它"的工作。
• 把免写前提写进 daemon 标志旁边的注释。 就像 enrich_nominations() 做的那样，把前提写在代码所在之处，让未来任何在 worker 里加磁盘写入的改动都会撞上警告。
• 显式测试 daemon-ness。 进程挂起类 bug 对结果导向测试不可见——那个通过的"慢话题"测试证明了错误的东西。在 worker 内断言 threading.current_thread().daemon（一行断言）就能把预算依赖的那个属性钉住。
• worker 内的 per-request 超时仍是第一道防线。 daemon 放弃只是针对病态情形的最后防线；worker 内每次网络调用仍应自带超时。
• 评审时，把 finally 里的 shutdown(wait=False, cancel_futures=True) 当作一个信号。 它是人们想要"放弃"语义时惯用的惯用法，但它并不提供"放弃"。

适用范围说明：这个模式只应用在可放弃掉队者的位置

一个容易被误读的细节值得单独说明：ThreadPoolExecutor 并没有被从代码库中移除。从源码结构看，它仍在其他位置被使用——例如 run() 内部对 subquery×source 的扇出（pipeline.py L2349-L2350），那里的工作是真正打算等待的；amazon.py L601-L610 甚至带有一段注释，解释自己为什么刻意不用 with 块（context manager 退出会 shutdown(wait=True)，会被"刚被截止期判死的掉队者"拖住）。

因此正确的使用姿势是文档原话的意思："应用掉队者问题，而不是应用这个模式。" 在改动 pipeline.py 其他 executor 调用点时，先问"这里的工作是否可以被放弃、被放弃时是否会损坏共享状态"，再决定保留 executor 还是切换到 daemon 线程方案。

相关学习记录

本缺陷与另外两条来自同一 PR #816 发现模式重构的经验互为姊妹篇，均落在 pipeline.py 中：

• 排序输出置信度下限与诚实的空状态——同一 PR 的"排序质量"半边，与本文的"进程生命周期"半边互补；
• argparse 可选值 flag 的分发真值陷阱——同一 PR 的第三课：bare-flag 与 flag-absent 的混淆，另一类对结果导向测试不可见的缺陷。

总结：在 Python 里给线程池加预算时，先区分你约束的是"答案"还是"进程"。as_completed(timeout=...) 与 shutdown(wait=False, cancel_futures=True) 只能做前者；要做到后者，必须让 worker 以 daemon 线程身份运行、以单调时钟截止期驱动消费者、以信号量保留并发上限，并且用测试把"worker 是守护线程"这一行属性直接钉死——这正是 last30days-skill 的 enrichment 预算从"幻觉"变成"合同"的完整路径。