同步工具超时后，后台线程为什么还在跑

把一个旧的同步读取函数接入 async Agent，最容易写出的包装是 await asyncio.wait_for(asyncio.to_thread(read), timeout)。调用方终于能及时收到超时，于是你让它开始下一次尝试。过了一会，旧读取又回来了：它可能仍占着线程池名额，仍使用着你准备关闭的客户端，还可能把旧值写进新一轮状态。

这篇要做的，是把这三个寿命分开验收：等待何时结束、实际工作何时结束、结果到哪一轮仍然有效。你会在 Python 标准库中运行三个门控实验，亲眼看到 Task 的取消状态与线程的完成状态分离，再把停止信号、资源归属和结果版本检查接起来。

需要会用 async/await 和 try/finally。实验实际运行于 Python 3.12.14，全部输入是内存合成值；线程、线程池和取消是真实运行的。没有模型、第三方 SDK、网络请求、数据库、密钥或外部业务操作。不要把这里的受控 worker 换成一个可能无限阻塞的未知库，再依赖最后的清理代码兜底。

第一个实验先创建 to_thread 的 Task，再等 worker 发出 started。这个顺序很重要：如果工作还在池里排队，取消确实可能让它根本不执行。本文观察的是已经运行的同步函数。worker 随后停在 threading.Event 的 release 门上；门最多等待两秒，失效时会报错。

调用方给 wait_for 10 毫秒预算，触发 TimeoutError 后断言 task.cancelled() 为真，同时 finished 尚未设置。接着 finally 打开 release 门，线程继续追加 local_work_after_timeout，然后发出 finished。这个本地列表变化就是“后台工作继续”的可观察证据，不涉及远端效果。

wait_for 会取消其等待对象，并等待取消完成，因此总耗时可能超过参数里的 timeout。to_thread 将同步函数放到另一线程。这里被取消的是 asyncio Task；Task 的终态不等于同步函数已经返回。

日志里的 cancelled_task_can_return=false 只表示这个已取消的 Task 不能再返回正常结果。它没有证明计算结果从未产生，也没有证明所有其他引用都失去了结果。下一组实验会主动保存底层 Future，继续观察其结局。

第二个实验使用专用 ThreadPoolExecutor.submit，保存 concurrent.futures.Future，再通过 asyncio.wrap_future 接到事件循环。wait_for 等的是 shield(wrapped)。shield 在这一个等待窗口里保护 wrapped 不被超时取消，使我们之后还能等到 worker 的结果；它不会终止线程，也不会替我们管理资源。

确认 future.running() 后，future.cancel() 实测返回 False。这符合 Future.cancel 对运行中任务的约定。随后调用方设置 stop Event。只有 worker 正在检查或等待这个 Event，停止请求才会得到响应。

BytesIO 在 worker 内创建，也由 worker 的 with 语句关闭。trace 先记录 observed_stop 时 closed=false，再记录 resource_closed 时 closed=true，最后才返回 stopped。若由外层协程在 timeout 的 finally 直接关掉资源，后台线程可能仍在使用它。应当让资源所有者负责关闭，或者另有可以确认所有使用者都结束的协调者。

示例把“观察停止”与“真正完成”都留下了记录。生产里的 stop.set() 只是请求，不能作为关闭客户端、归还名额或宣布已停机的证据。停止确认需要明确的完成信号，必要时还要保留底层 Future 的异常。

第三个实验把 attempt 1 放进 max_workers=1 的专用池，等待它开始后再提交一个排队项。旧调用超时以后，first 仍 running，queued 既没有运行也没有完成。另一个 asyncio.to_thread 调用走默认池，仍能返回 healthy。这个隔离可以避免某类慢工具直接占满默认池。

然后执行 shutdown(wait=False, cancel_futures=True)。排队项进入 cancelled，运行中的 first 留下继续执行。官方 shutdown 约定同样说明，wait=False 不代表整个 Python 程序能立刻退出；尚在运行的线程仍需结束。

专用池隔离的是容量，不是权限、内存或进程故障。max_workers 也不限制你向池里提交多少项。持续接收输入时还要有有界入场规则，不能把线程数上限当成队列上限。如果外层 Semaphore 在 Task 超时时就释放，而底层线程仍在工作，这个 Semaphore 统计的也不再是真实运行数。

等待旧 worker 时，主线程把 current_attempt 更新为 2，状态已有 value=new。释放旧 worker 后，它返回 attempt=1 和 value=old。accept_result 对比 attempt 后拒绝写入，state 继续保留 new。这是应用层结果有效期检查，不是 Python 自动提供的行为。

本例只处理一个逻辑读取，attempt 在这个读取范围内递增。真正同时管理多个调用时，至少要把稳定的调用身份与 epoch 一起校验，并从可信调度器分配版本，不能让工具自己宣称“我属于当前版本”。示例不接受不完整结果；缺失字段会抛出 KeyError，交给测试暴露。

比较与写入只在事件循环线程中执行，中间没有 await。若多个线程或进程同时更新状态，需要相应的锁或数据库条件更新，把版本比较和写入放到同一个原子边界。本文的简单函数不能替代跨进程并发控制。即使结果被拒绝，旧工具已经发生的外部操作也不会因此消失。

新建一个目录，保存下面的 timeout_lab.py。这里只依赖标准库，不需要 pip install。三个实验的全部输入都在函数内：10 毫秒调用方等待预算、两秒门控保护、BytesIO 的 local fixture、旧结果 old、新状态 new，以及 attempt 1 和 2。两秒保护是实验失效上限，不是生产建议值。

event() 用异步短暂停顿观察线程事件，没有再占一个线程去等待另一个线程。每个 finally 都会发出释放或停止信号。专用池的 shutdown(wait=True) 负责 join，默认池在线程门释放后由 asyncio.run 的收尾等待结束。最后枚举线程，要求没有主线程以外的残留。

timeout_lab.py · 完整文件

"""Bounded local evidence only. Python standard library; no external effects."""
import asyncio
import concurrent.futures as cf
import io
import json
import platform
import threading

async def event(event, seconds=2):
    # No thread-pool work to observe a thread-pool worker.
    async with asyncio.timeout(seconds):
        while not event.is_set():
            await asyncio.sleep(.001)

def emit(name, **values):
    print(json.dumps(dict(experiment=name, **values), sort_keys=True), flush=True)

async def timeout_is_not_exit():
    started, release, finished = (threading.Event() for _ in range(3))
    trace = []
    def worker():
        started.set()
        try:
            if not release.wait(2):
                raise RuntimeError('safety gate expired')
            trace.append('local_work_after_timeout')
            return 'old-result'
        finally:
            finished.set()
    task = asyncio.create_task(asyncio.to_thread(worker))
    try:
        await event(started)
        try:
            await asyncio.wait_for(task, .01)
            raise AssertionError('expected timeout')
        except TimeoutError:
            pass
        assert task.cancelled() and not finished.is_set()
        emit('1_timeout_is_not_exit', task_cancelled=task.cancelled(),
             worker_finished=finished.is_set(), cancelled_task_can_return=False)
    finally:
        release.set()
        await event(finished)
        await asyncio.gather(task, return_exceptions=True)
    assert trace == ['local_work_after_timeout']
    emit('1_cleanup', worker_finished=finished.is_set(), trace=trace)

async def cooperative_owner():
    started, stop, finished = (threading.Event() for _ in range(3))
    trace = []
    pool = cf.ThreadPoolExecutor(max_workers=1, thread_name_prefix='cooperative')
    def worker():
        # Worker owns its resource through its full physical lifetime.
        with io.BytesIO(b'local fixture') as resource:
            started.set()
            if not stop.wait(2):
                raise RuntimeError('stop was never signalled')
            trace.append(('observed_stop', resource.closed))
        trace.append(('resource_closed', resource.closed))
        finished.set()
        return 'stopped'
    future = pool.submit(worker)
    wrapped = asyncio.wrap_future(future)
    try:
        await event(started)
        try:
            await asyncio.wait_for(asyncio.shield(wrapped), .01)
            raise AssertionError('expected timeout')
        except TimeoutError:
            pass
        assert future.running() and not finished.is_set()
        assert future.cancel() is False
        stop.set()
        result = await asyncio.wait_for(asyncio.shield(wrapped), 2)
        assert result == 'stopped' and trace == [('observed_stop', False), ('resource_closed', True)]
        emit('2_cooperative_owner', running_cancel_return=False,
             stop_acknowledged=finished.is_set(), trace=trace)
    finally:
        stop.set()
        pool.shutdown(wait=True, cancel_futures=True)

def accept_result(current_attempt, state, result):
    # Call only on the event-loop thread, without an await between check and write.
    if result['attempt'] != current_attempt:
        return False
    state['value'] = result['value']
    return True

async def isolation_and_stale_result():
    started, release, finished, queued_ran = (threading.Event() for _ in range(4))
    pool = cf.ThreadPoolExecutor(max_workers=1, thread_name_prefix='slow-tool')
    def old_attempt():
        started.set()
        try:
            if not release.wait(2):
                raise RuntimeError('safety gate expired')
            return {'attempt': 1, 'value': 'old'}
        finally:
            finished.set()
    first = pool.submit(old_attempt)
    wrapped = asyncio.wrap_future(first)
    try:
        await event(started)
        queued = pool.submit(queued_ran.set)
        try:
            await asyncio.wait_for(asyncio.shield(wrapped), .01)
            raise AssertionError('expected timeout')
        except TimeoutError:
            pass
        # Capacity is still occupied after the caller deadline. Default pool is independent.
        assert first.running() and not queued.running() and not queued.done()
        independent = await asyncio.wait_for(asyncio.to_thread(lambda: 'healthy'), 2)
        assert independent == 'healthy'
        # Shutdown cancels queue entries, not the currently running callable.
        pool.shutdown(wait=False, cancel_futures=True)
        assert queued.cancelled() and first.running()
        current_attempt, state = 2, {'value': 'new'}
        release.set()
        late = await asyncio.wait_for(asyncio.shield(wrapped), 2)
        accepted = accept_result(current_attempt, state, late)
        assert not accepted and state['value'] == 'new' and not queued_ran.is_set()
        emit('3_isolation_and_stale_result', queue_cancelled=queued.cancelled(),
             independent_pool=independent, late_result_accepted=accepted, state=state,
             worker_finished=finished.is_set())
    finally:
        release.set()
        pool.shutdown(wait=True, cancel_futures=True)

async def main():
    await timeout_is_not_exit()
    await cooperative_owner()
    await isolation_and_stale_result()

if __name__ == '__main__':
    print('Python', platform.python_version(), flush=True)
    asyncio.run(main())
    alive = [t.name for t in threading.enumerate() if t is not threading.main_thread()]
    assert not alive, alive
    emit('final', remaining_threads=alive)

运行 python3 timeout_lab.py。下面是本次 Python 3.12.14 的实际标准输出。事件门决定关键先后，10 毫秒只负责触发真实等待超时；程序没有验收“恰好在第 10 毫秒返回”。

Python 3.12.14
{"cancelled_task_can_return": false, "experiment": "1_timeout_is_not_exit", "task_cancelled": true, "worker_finished": false}
{"experiment": "1_cleanup", "trace": ["local_work_after_timeout"], "worker_finished": true}
{"experiment": "2_cooperative_owner", "running_cancel_return": false, "stop_acknowledged": true, "trace": [["observed_stop", false], ["resource_closed", true]]}
{"experiment": "3_isolation_and_stale_result", "independent_pool": "healthy", "late_result_accepted": false, "queue_cancelled": true, "state": {"value": "new"}, "worker_finished": true}
{"experiment": "final", "remaining_threads": []}

第一行实验记录同时出现 task_cancelled=true 和 worker_finished=false；下一行证明线程后来继续工作并完成。第二组中的两个 closed 值确认关闭顺序。第三组同时确认排队取消、默认池可用与旧值拒绝，末尾 remaining_threads=[] 只在 asyncio.run 完成收尾之后打印。

如果 safety gate expired、stop was never signalled 或观察事件超时报错，先检查本地环境负载和修改过的门控顺序。不要把保护上限改成无限等待，也不要删掉断言来让测试变绿。

将以下独立写下的预期保存为 expected.json。它不含 Python 版本行，也不含机器相关的耗时。测试逐个比较 JSON 对象，不依赖键在文本中的顺序；不要从一次运行日志重新生成这个文件。

expected.json · 完整文件

[
  {"experiment":"1_timeout_is_not_exit","cancelled_task_can_return":false,"task_cancelled":true,"worker_finished":false},
  {"experiment":"1_cleanup","trace":["local_work_after_timeout"],"worker_finished":true},
  {"experiment":"2_cooperative_owner","running_cancel_return":false,"stop_acknowledged":true,"trace":[["observed_stop",false],["resource_closed",true]]},
  {"experiment":"3_isolation_and_stale_result","independent_pool":"healthy","late_result_accepted":false,"queue_cancelled":true,"state":{"value":"new"},"worker_finished":true},
  {"experiment":"final","remaining_threads":[]}
]

再保存 test_timeout_lab.py，运行 python3 -m unittest -v。端到端测试在独立子进程运行三个实验，十五秒后仍未退出就失败；subprocess.run 的 timeout 会终止这个只含本地合成工作的子进程。正常通过要求退出码为零、无 stderr、全部预期匹配，并包含无残留线程记录。

其余五项分别检验当前版本可写、旧版本拒绝、其他版本拒绝、缺失版本字段报错，以及事件观察本身有时限。正向用例能防止把 accept_result 偷懒写成永远返回 False。

test_timeout_lab.py · 完整文件

"""Independent expectations; no regeneration of expected.json from actual output."""
import asyncio
import json
from pathlib import Path
import subprocess
import sys
import threading
import unittest
from timeout_lab import accept_result, event

ROOT = Path(__file__).resolve().parent

class TimeoutLabTests(unittest.TestCase):
    def test_three_experiments_and_thread_join(self):
        run = subprocess.run(
            [sys.executable, str(ROOT / 'timeout_lab.py')],
            cwd=ROOT, capture_output=True, text=True, timeout=15,
        )
        self.assertEqual(run.returncode, 0, run.stderr)
        self.assertEqual(run.stderr, '')
        lines = run.stdout.splitlines()
        self.assertTrue(lines[0].startswith('Python '))
        actual = [json.loads(line) for line in lines[1:]]
        expected = json.loads((ROOT / 'expected.json').read_text())
        self.assertEqual(actual, expected)

    def test_current_attempt_is_accepted(self):
        state = {'value': 'before'}
        self.assertTrue(accept_result(2, state, {'attempt': 2, 'value': 'current'}))
        self.assertEqual(state, {'value': 'current'})

    def test_old_attempt_cannot_overwrite(self):
        state = {'value': 'new'}
        self.assertFalse(accept_result(2, state, {'attempt': 1, 'value': 'old'}))
        self.assertEqual(state, {'value': 'new'})

    def test_other_epoch_cannot_overwrite(self):
        state = {'value': 'new'}
        self.assertFalse(accept_result(2, state, {'attempt': 3, 'value': 'other'}))
        self.assertEqual(state, {'value': 'new'})

    def test_missing_result_fields_are_not_success(self):
        with self.assertRaises(KeyError):
            accept_result(2, {'value': 'new'}, {'value': 'missing epoch'})

    def test_observer_has_timeout(self):
        async def check():
            with self.assertRaises(TimeoutError):
                await event(threading.Event(), .01)
        asyncio.run(check())

if __name__ == '__main__':
    unittest.main()

test_current_attempt_is_accepted (test_timeout_lab.TimeoutLabTests.test_current_attempt_is_accepted) ... ok
test_missing_result_fields_are_not_success (test_timeout_lab.TimeoutLabTests.test_missing_result_fields_are_not_success) ... ok
test_observer_has_timeout (test_timeout_lab.TimeoutLabTests.test_observer_has_timeout) ... ok
test_old_attempt_cannot_overwrite (test_timeout_lab.TimeoutLabTests.test_old_attempt_cannot_overwrite) ... ok
test_other_epoch_cannot_overwrite (test_timeout_lab.TimeoutLabTests.test_other_epoch_cannot_overwrite) ... ok
test_three_experiments_and_thread_join (test_timeout_lab.TimeoutLabTests.test_three_experiments_and_thread_join) ... ok

----------------------------------------------------------------------
Ran 6 tests in 0.174s

OK

先查工具自己的连接、读取和总操作时限。一个阻塞库如果不提供有效超时或中断 API，外层 stop Event 无法抢占它；Python threading 接口也没有安全强杀任意运行线程的通用能力。对必须具有硬终止边界的工作，应另行设计可回收的进程或外部 worker，核验终止后的数据与资源状态。本文没有实现或验证进程隔离。

本地演示的 finally 里能同步 shutdown(wait=True)，是因为所有 worker 都受两秒门控制，且 finally 已经打开门。把相同 shutdown 塞进真实 async 请求处理器，可能阻塞事件循环并超过请求期限。生产服务需要由请求之外的生命周期管理者持有池、追踪运行项、接收迟到结果或异常，并落实退出策略；本页不是可直接拷进生产的通用工具适配器。

还应明确超时后是否允许重试、重试是否共享客户端，以及旧工作完成前是否能归还真实容量。请求期限到点可以停止等待；如果系统承诺线程已退出或资源已关闭，就必须有另一份完成证据。保留这种区别，才能解释为什么用户已经收到超时，后台监控里仍有活跃工具。

本实验没有覆盖多次外部取消、真实阻塞网络库、CPU 密集代码、生产吞吐或进程终止。它没有使用 TaskGroup，也不依赖其他文章对 TaskGroup 取消竞争的结论。

运行版本为 Python 3.12.14；官方资料核对日期为 2026 年 10 月 7 日，所查 Python 3.12 页面标示 3.12.15。本文使用的 timeout 需要 Python 3.11 以上，但没有实测其他版本，不据此承诺相同平台行为。

API 依据：asyncio.shield、asyncio.to_thread、Future.cancel、Executor.shutdown。停止协议、资源归属、epoch 与 expected.json 都是本文的应用设计。

Python 文档许可页说明软件和文档使用 PSF License Version 2，文档代码示例另外提供 Zero-Clause BSD 许可。本文实验代码和三张 SVG 为独立编写，没有复制官方示例或第三方图片。

Agent Loop讨论模型循环与调用管理；工具检查时序讨论检查、提交和回执。这一页只补同步工具移到后台线程后的寿命问题：协程已经退出，真实工作却可能仍持有资源。