一次 aiosqlite 警告背后的资源生命周期漏洞
一次 pytest 非致命 aiosqlite 告警的完整复盘：从偶发的事件循环关闭报错，定位到 ConversationManager 未持有并关闭 checkpoint 连接，再用事务式启动、逆序释放和测试夹具修正资源所有权。
这不是一条可以忽略的测试告警
全量测试原本是绿的，但结束时总会跟着一条 PytestUnhandledThreadExceptionWarning：aiosqlite 的后台线程试图把结果投递回一个已经关闭的 asyncio 事件循环。
它没有让 CI 失败，也没有在业务路径上立刻报错，所以很容易被归类为测试环境噪声。可这类告警有一个不应回避的事实：有一段后台工作活到了它所属的事件循环之后。把 warning 过滤掉，只是让资源泄漏更安静，并没有让它消失。
这次排查的目标因此不是“让 pytest 不显示 warning”，而是回答三个更具体的问题：哪个对象创建了后台线程，谁应当关闭它，启动走到一半失败时又由谁收尾。
先确认：get_history 只是告警出现的位置
warning 最常落在 test_get_history_thread_isolation 附近。最开始看起来很像历史查询触发了问题，但这个判断经不起最小验证。
ConversationManager.start() 会打开一个 aiosqlite 连接，把它交给 LangGraph 的 AsyncSqliteSaver 作为 checkpoint 存储；测试执行完后，manager 却没有关闭这个连接。查询历史只是恰好让测试在那个时刻结束，随后 pytest 开始销毁事件循环，遗留线程才暴露出来。
我没有继续沿着 get_history() 的查询逻辑猜，而是直接检查连接工作线程的生存状态：
before_stop_worker_alive=True
after_stop_worker_alive=True
after_direct_close_worker_alive=False这组结果把问题收窄得很干净：调用旧的 ConversationManager.stop() 后，checkpoint 连接的 worker 仍在；直接 await connection.close() 后，线程才退出。于是“查询隔离有 bug”的假设可以排除，根因是生命周期没有闭合。
这也解释了为什么它是非致命的。aiosqlite 为每个连接创建独立的 worker thread，操作通过队列回到创建连接的事件循环。正常的 close() 会完成队列中的工作、关闭 SQLite 连接，再向 worker 发送停止信号；若等到解释器或 pytest 在事件循环关闭后才被动清理，线程就可能再尝试回调那个循环。aiosqlite 的连接实现里可以看到 worker、close() 和析构保护是同一条链路。
根因不在库，而在所有权被拆开了
把提交历史和当前代码放在一起看，问题的形成过程很明确。
最早的 checkpoint 接入创建了连接，并立即把它包装成 AsyncSqliteSaver：
conn = await aiosqlite.connect(self._db_path)
self._checkpointer = AsyncSqliteSaver(conn)之后，为对话摘要增加 memory database 时，ConversationManager.stop() 被补上，但它只知道 self._memory_db。checkpoint 连接没有保存在 manager 上，只有 saver 间接持有它；而 saver 不是连接的创建者，也没有被赋予关闭连接的职责。
问题不是“少写了一句 close”这么简单。它违反了一条很朴素的资源规则：谁创建资源，谁就必须拥有明确、可达、可测试的释放路径。将连接藏进包装对象后，关闭责任也一起丢了。
设计文档把这件事写成了一个很小的所有权表：
| 资源 | 使用者 | 明确所有者 |
|---|---|---|
| checkpoint SQLite 连接 | AsyncSqliteSaver | ConversationManager._checkpoint_db |
| memory SQLite 连接 | 摘要读写方法 | ConversationManager._memory_db |
这里刻意没有通过 AsyncSqliteSaver 的私有字段反向挖连接。包装器负责 checkpoint 协议，manager 负责自己打开的数据库连接；各自边界清楚，之后换 saver 或调整内部实现也不会改变关闭职责。
方案取舍：修资源，不修表象
排查后有过几种看起来更快的做法：在 pytest 里过滤 warning、给单个测试手动补 close()、或者在 __del__ 里兜底。它们都不满足这次的验收条件。
- 过滤 warning 掩盖了线程没有停止的事实。
- 只改出问题的测试，服务进程的正常停止路径仍然泄漏。
- 用析构函数兜底时，事件循环可能早已关闭；库源码也明确把它当作最后防线，而不是正常生命周期。
- 让 saver 的内部对象承担清理，会把
ConversationManager的资源所有权隐含在第三方实现细节里。
最终选择的是最窄的改造：不替换 SQLite，不引入连接池，也不做通用“资源管理框架”。只让 ConversationManager 显式拥有它已经创建的两个连接，并把启动和停止做成对称的生命周期。
把启动当成一个小事务
正常关闭很好理解，真正容易漏的是启动中途失败。例如 checkpoint 已经打开，但 memory database 的建表或迁移失败。如果字段在每一步完成后立刻写入实例，异常路径很容易留下半初始化对象，也容易忘记关闭较早打开的连接。
改造后的 start() 先只使用局部变量：
checkpoint_db = None
memory_db = None
try:
checkpoint_db = await aiosqlite.connect(self._db_path)
checkpointer = AsyncSqliteSaver(checkpoint_db)
await checkpointer.setup()
graph_compiled = self._build_graph().compile(checkpointer=checkpointer)
if self._memory_db_path:
memory_db = await aiosqlite.connect(self._memory_db_path)
await memory_db.execute("CREATE TABLE IF NOT EXISTS memories ...")
except BaseException:
for connection in (memory_db, checkpoint_db):
if connection is not None:
await connection.close()
raise
self._checkpoint_db = checkpoint_db
self._checkpointer = checkpointer
self._graph_compiled = graph_compiled
self._memory_db = memory_db这段代码有两个刻意的约束。
第一，实例状态只在所有初始化成功后才发布。调用方看不到“有 saver、没有 graph”或“有 checkpoint、memory 初始化了一半”的对象。
第二，失败时按创建的反方向关闭。memory 是后创建的，先关 memory；checkpoint 先创建，最后关 checkpoint。每个关闭都有自己的异常保护，因此清理失败不会覆盖原始启动异常。这里捕获 BaseException，是为了在取消等非 Exception 的退出路径中也不遗留已经打开的连接，然后仍把原异常原样抛回调用方。
它和数据库事务很像：全部准备成功才提交到对象状态；任何一步失败，就回滚已经取得的外部资源。不同的是，回滚对象不是一行数据，而是一个带后台线程的异步连接。
停止顺序、幂等性和状态清空
stop() 的职责同样被收敛为唯一入口。它先把字段复制到局部变量，再立刻把实例字段置空，随后关闭 memory 和 checkpoint：
async def stop(self) -> None:
memory_db = self._memory_db
checkpoint_db = self._checkpoint_db
self._memory_db = None
self._checkpoint_db = None
self._checkpointer = None
self._graph_compiled = None
try:
if memory_db is not None:
await memory_db.close()
finally:
if checkpoint_db is not None:
await checkpoint_db.close()先清字段并不是为了省几行代码，而是让第二次调用成为自然的空操作。关闭 memory 时即使抛异常，finally 仍会关闭 checkpoint，保证后创建资源的失败不会阻断较早资源的回收。
服务的停止入口原本就调用 await conversation.stop()，因此生产路径不需要新增 API。改变的是 stop() 终于与 start() 对称：打开什么，就关闭什么；构建什么，就让对象状态回到未启动状态。
测试不该绕过生命周期
修复只加一条“调用 stop()”的测试还不够。原测试里很多地方直接 await manager.start()，断言结束后并不保证释放连接；即使业务测试都通过，测试自身仍可能留下后台资源。
于是把 manager 的创建集中为 async yield fixture：
@pytest_asyncio.fixture
async def conversation_manager_factory():
managers = []
async def create(chat_model, **kwargs):
manager = ConversationManager(db_path=":memory:", chat_model=chat_model, **kwargs)
await manager.start()
managers.append(manager)
return manager
try:
yield create
finally:
for manager in reversed(managers):
await manager.stop()这样测试的正常返回、断言失败和中途异常都会进入 finally。工厂还能覆盖一个测试创建多个 manager 的情形，并按反向创建顺序关闭它们。直接构造仅保留给两种不打开资源的场景，或专门验证启动失败回滚的场景。
新增的回归测试不是只检查字段为 None，而是检查 stop() 后 checkpoint worker thread 已经退出，再次 stop() 也不会报错。另一条测试让 memory 数据库初始化故意抛出 OSError，验证此前成功打开的 checkpoint 和刚打开的 memory 都被关闭，同时所有生命周期字段保持未赋值。它把“启动中途失败也不泄漏连接”变成可执行的验收条件。
最终结果和可复用的判断方法
这次改造落在两个提交中：先用设计文档固定所有权、启动回滚、逆序关闭和测试标准，再实现 ConversationManager 与统一测试夹具。设计提交 019210b 和 修复提交 b28b1bf 分别保留了决策与代码证据。
最终执行 uv run pytest 得到 454 passed in 13.19s，此前的 PytestUnhandledThreadExceptionWarning 没有再出现。这里的通过不是因为 warning 被压制，而是因为测试循环关闭之前，连接自己的 worker 已经被正常停止。
这类问题以后可以用一组简单的问题快速判断：
- 这个对象打开了哪些外部资源，包括线程、连接、文件和子进程？
- 每项资源由哪个对象明确拥有，释放入口是否可达？
- 初始化失败在第 N 步时，前 N-1 项是否都会按逆序回收？
- 测试是否通过真实生命周期创建资源，并保证 teardown 一定运行？
非致命 warning 往往不是“不影响功能”的同义词。它更像一个早期信号：程序的资源边界还没有被写完整。越早沿着创建、所有权和释放三件事追下去，越不需要在生产环境里用一次卡住的关机或偶发的线程异常来补课。
Sources
- aiosqlite Connection 源码 — worker thread、close() 和未关闭连接的保护逻辑。
- LangGraph SQLite checkpoint 实现 — AsyncSqliteSaver 所在的官方实现。
- 019210b: ConversationManager resource lifecycle design — 本次所有权与验收标准的设计记录。
- b28b1bf: own ConversationManager database lifecycle — 本次实现、失败回滚和测试夹具改造。