fix(async): 把事件循环上的同步落盘收口，并给这一类加 CI 守卫

brain/task_executor.py

@@ -427,7 +427,18 @@ async def _prewarm_short_descriptions(
         finally:
             self._short_desc_prewarm_inflight -= pids
             # 把本批生成的（贵的）条目落盘，下次启动直接复用、不再现生成。
-            self._persist_generated_short_descriptions(generated)
+            # 这里刻意保留同步落盘（不改 await asyncio.to_thread），两个原因：
+            # 1) _persist_generated_short_descriptions 内部是「读盘—合并—写盘」，
+            #    全程没有锁；今天靠「整段同步、不让出事件循环」才保证两批并发
+            #    prewarm 不互相覆盖（见该函数里 re-read 那行注释）。挪进线程后，
+            #    两批会各自在自己的 worker 线程里 load→merge→write 交错，先写的
+            #    那批条目会被后写的整份 payload 盖掉。
+            # 2) 这是 finally，而本协程绝大部分时间挂在 llm.ainvoke 上——事件循环
+            #    收尾时它正是会被 cancel 的 pending task。在取消路径的 finally 里
+            #    await，落盘可能被直接跳过，白白丢掉花了 LLM 调用生成的条目。
+            # 代价可控：每批 prewarm 只写一次小 JSON，发生在插件加载期，不在
+            # analyze 热路径上。
+            self._persist_generated_short_descriptions(generated)  # noqa: ASYNC_BLOCK — 无锁读-改-写 + 取消路径 finally，加 await 会引入互相覆盖/漏落盘
 
     async def plugin_list_provider(self, force_refresh: bool = True) -> List[Dict[str, Any]]:
         # return cached list when allowed

main_logic/core/proactive.py

@@ -487,22 +487,6 @@ async def finish_proactive_delivery(
                 # 的台词天生模板化，录进 corpus 会污染 FG 窗、漂移其它 channel 的
                 # 复读基线，故按 ANTI_REPEAT_EXEMPT_SOURCE_TAGS 豁免（与出口的
                 # BM25 评分豁免对偶）。
-                if source_tag not in ANTI_REPEAT_EXEMPT_SOURCE_TAGS:
-                    try:
-                        from memory.anti_repeat import get_anti_repeat_corpus
-                        get_anti_repeat_corpus().record_output(
-                            self.lanlan_name,
-                            full_text,
-                            is_proactive=True,
-                            now=(
-                                publication_times[0]
-                                if publication_times
-                                else None
-                            ),
-                        )
-                    except Exception as _exc:  # pragma: no cover
-                        logger.debug("[AntiRepeat] record proactive skipped: %s", _exc)
-
             if self.use_tts and self.tts_thread and self.tts_thread.is_alive() and not self._tts_done_queued_for_turn:
                 try:
                     await self._request_tts_done_for_turn("finish_proactive_delivery")
@@ -519,6 +503,29 @@ async def finish_proactive_delivery(
             except Exception:
                 # Turn-end push is best-effort; the client may have gone away.
                 pass
+
+            # 防复读 corpus 排在**所有收尾信号之后**。落盘走 async 孪生（同步版
+            # 尾部是 atomic_write_json，含无上界 fsync，压在会话循环上就是掐音
+            # 频），但那个 await 也是个取消点：文本此刻已经投递出去了，被取消的
+            # 话 CancelledError 是 BaseException、下面的 except Exception 接不住，
+            # TTS 收尾和两处 turn end 就全被跳过 —— 用户看得见的一轮没有终止信号，
+            # 比漏录一条语料严重得多。与 omni_offline_client/_lifecycle.py 同款处置。
+            #
+            # LLM 给自己的元数据备忘，不算复读对象。素材推送类 channel（推歌）
+            # 的台词天生模板化，录进 corpus 会污染 FG 窗、漂移其它 channel 的
+            # 复读基线，故按 ANTI_REPEAT_EXEMPT_SOURCE_TAGS 豁免（与出口的
+            # BM25 评分豁免对偶）。
+            if source_tag not in ANTI_REPEAT_EXEMPT_SOURCE_TAGS:
+                try:
+                    from memory.anti_repeat import get_anti_repeat_corpus
+                    await get_anti_repeat_corpus().arecord_output(
+                        self.lanlan_name,
+                        full_text,
+                        is_proactive=True,
+                        now=publication_times[0] if publication_times else None,
+                    )
+                except Exception as _exc:  # pragma: no cover
+                    logger.debug("[AntiRepeat] record proactive skipped: %s", _exc)
         # proactive 原文不写 logger（隐私）；本地 print 兜底
         logger.info("[%s] Proactive stream delivered (text_len=%d)", self.lanlan_name, len(full_text or ""))
         print(f"[{self.lanlan_name}] Proactive stream delivered: {(full_text or '')[:40]}…")

main_logic/omni_offline_client/_lifecycle.py

@@ -432,22 +432,31 @@ async def prompt_ephemeral(
                     logger.exception("prompt_ephemeral on_committed callback failed")
             if content_committed and persist_response:
                 self._conversation_history.append(AIMessage(content=assistant_message))
+            if completion_mode == "response":
+                if self.on_response_done:
+                    await self.on_response_done()
                 # 防复读 corpus：只录常规 reply（completion_mode == "response"）。
                 # proactive 路径已经在 ``core.finish_proactive_delivery`` 上录，
                 # 这里再录会双写——这两条路径都接得到同一段 assistant 文本。
-                if completion_mode == "response":
+                #
+                # ⚠️ 必须排在 on_response_done **之后**。落盘走 async 孪生（同步版
+                # 尾部是 atomic_write_json，含无上界的 fsync，而这条路径每条回复都
+                # 走一次，压在会话循环上就是掐音频），但那个 await 也就成了一个取消
+                # 点：文本已经提交、历史已经写上，此时被取消的话 CancelledError 是
+                # BaseException，下面的 except Exception 接不住，on_response_done
+                # 里的 TTS 收尾 / turn 结束 / request-id 清理就全被跳过 —— 一次
+                # 已提交的回复没有终止信号，比漏录一条防复读语料严重得多。
+                # 排在后面，收尾信号先落地，corpus 只是尽力而为。
+                if content_committed and persist_response:
                     try:
                         from memory.anti_repeat import get_anti_repeat_corpus
-                        get_anti_repeat_corpus().record_output(
+                        await get_anti_repeat_corpus().arecord_output(
                             self.lanlan_name, committed_text, is_proactive=False,
                         )
                     except Exception as _exc:  # pragma: no cover
                         logger.debug(
                             "[AntiRepeat] record reply skipped: %s", _exc,
                         )
-            if completion_mode == "response":
-                if self.on_response_done:
-                    await self.on_response_done()
             else:
                 proactive_done_cb = getattr(self, "on_proactive_done", None)
                 if proactive_done_cb:

main_routers/storage_location_router.py

@@ -86,6 +86,30 @@
 logger = logging.getLogger(__name__)
 _storage_mutation_lock = asyncio.Lock()
 
+# _STORAGE_MUTATION_STAYS_ON_LOOP
+#
+# 本文件里所有写存储状态的调用**刻意**留在事件循环上（各处 noqa: ASYNC_BLOCK 指向
+# 这里）。落盘本身是同步的、带无上界的 fsync，按理该挪进 to_thread —— 但这里有两条
+# 独立的理由压过它，任缺其一都会造成比循环卡顿严重得多的后果。
+#
+# 一、这些写是**取消原子**的序列，而它们之间今天一个 await 都没有。
+#     典型形状：delete_storage_migration → save_storage_policy → set_root_mode。
+#     任何一处插进 await，请求超时或应用关闭产生的 CancelledError 就能落在中间：
+#     恢复检查点已删、策略已写，而 root mode 还是旧值。CancelledError 是
+#     BaseException，外层 `except Exception` 接不住，也就没人回滚 —— 用户会得到
+#     一个「恢复闸自相矛盾」的盘上状态。
+#
+# 二、root_state 有一个**不在锁里**的写者。
+#     build_storage_location_bootstrap_payload → _reconcile_legacy_cleanup_pending_
+#     root_state（utils/storage/location_bootstrap.py:191）会 save_root_state，它挂在
+#     GET /bootstrap、/status、/diagnostics、/retained-source 和 POST /exit 上，这些
+#     都不在 _storage_mutation_lock 覆盖下。今天让「GET 侧 reconcile」和「变更路由的
+#     写」互斥的不是锁，是**它们都跑在同一条事件循环线程上**。把任何一个 root_state
+#     写者挪进 worker，前端存储页每 500ms 的 /status 轮询就能把它整份盖掉。
+#
+# 真正的收口是给 root_state 一把真锁，并让 GET 路由别在读路径上写盘。在那之前，
+# 这个文件宁可让罕见的存储变更请求同步落盘。
+
 
 class StorageLocationSelectionRequest(BaseModel):
     selected_root: str = Field(..., min_length=1, max_length=4096)
@@ -234,7 +258,23 @@ async def _release_storage_startup_barrier_or_rollback(
         await _release_storage_startup_barrier_if_needed(reason=reason)
     except Exception:
         try:
-            _restore_storage_mutation_state(config_manager, snapshot, anchor_root=anchor_root)
+            # 这一处**刻意**留在事件循环上。_restore_storage_mutation_state 的最后
+            # 一步是 config_manager.save_root_state()，而 root_state 还有另一个写者：
+            # build_storage_location_bootstrap_payload → _reconcile_legacy_cleanup_
+            # pending_root_state（utils/storage/location_bootstrap.py:191）也会
+            # save_root_state，它挂在 GET /bootstrap、/status、/diagnostics、
+            # /retained-source 和 POST /exit 上 —— 这几条**都不在
+            # _storage_mutation_lock 覆盖下**（锁只包 cleanup / select / restart 三条）。
+            #
+            # 今天让这两个「读 root_state — 改 — 写回」互斥的，不是锁，而是「它们都跑在
+            # 同一条事件循环线程上」。把回滚搬进 worker 就恰好打破这个不变量：前端存储页
+            # 每 500ms 轮询 /status，回滚写 root_state 的同时那边正拿着读到的旧 dict 往
+            # 回写，回滚会被整份盖掉 —— 迁移检查点和策略回滚了、root_state 没有，下次启动
+            # recovery_required 直接算成 False，恢复闸被跳过。
+            #
+            # 正确的收口是给 root_state 一把真锁、并让 GET 路由别在读路径上写盘，那是
+            # 独立的一份工作。在那之前，宁可让这条罕见的回滚路径同步落盘。
+            _restore_storage_mutation_state(config_manager, snapshot, anchor_root=anchor_root)  # noqa: ASYNC_BLOCK — 末步 save_root_state 与无锁 GET 路由的 root_state 读改写互斥，只靠「同在循环线程」保证
         except Exception:
             logger.exception(
                 "failed to rollback storage mutation state after startup barrier release failed",
@@ -1307,7 +1347,7 @@ async def _post_storage_location_retained_source_cleanup_locked(
         updated_payload["retained_source_mode"] = "cleaned"
         updated_payload["updated_at"] = _utc_now_iso()
         updated_payload["cleanup_completed_at"] = _utc_now_iso()
-        save_storage_migration(config_manager, updated_payload, anchor_root=anchor_root)
+        save_storage_migration(config_manager, updated_payload, anchor_root=anchor_root)  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
 
     try:
         root_state = config_manager.load_root_state()
@@ -1404,7 +1444,10 @@ async def _post_storage_location_select_locked(
 
                 state_snapshot = _snapshot_storage_mutation_state(config_manager, anchor_root=anchor_root)
                 delete_storage_migration(config_manager, anchor_root=anchor_root)
-                policy_payload = save_storage_policy(
+                # 整个 _post_storage_location_select_locked 都跑在 _storage_mutation_lock 里，
+                # 下面紧跟着的 set_root_mode / 解除启动闸也已经是 await，多一个让出点不改变
+                # 「失败即整体回滚 state_snapshot」的语义。
+                policy_payload = save_storage_policy(  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
                     config_manager,
                     selected_root=current_root,
                     selection_source=payload.selection_source,
@@ -1439,7 +1482,7 @@ async def _post_storage_location_select_locked(
                 }
 
             state_snapshot = _snapshot_storage_mutation_state(config_manager, anchor_root=anchor_root)
-            policy_payload = save_storage_policy(
+            policy_payload = save_storage_policy(  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
                 config_manager,
                 selected_root=current_root,
                 selection_source=payload.selection_source,
@@ -1473,7 +1516,7 @@ async def _post_storage_location_select_locked(
                 "selection_source": policy_payload["selection_source"],
             }
         state_snapshot = _snapshot_storage_mutation_state(config_manager, anchor_root=anchor_root)
-        policy_payload = save_storage_policy(
+        policy_payload = save_storage_policy(  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
             config_manager,
             selected_root=current_root,
             selection_source=payload.selection_source,
@@ -1731,7 +1774,9 @@ async def _post_storage_location_restart_locked(
         state_snapshot = _snapshot_storage_mutation_state(config_manager, anchor_root=anchor_root)
         try:
             delete_storage_migration(config_manager, anchor_root=anchor_root)
-            save_storage_policy(
+            # 同样在 _storage_mutation_lock 里；这一段本来就以 await _request_app_shutdown
+            # 收尾，多出来的让出点仍被同一个 try/except 的 state_snapshot 回滚覆盖。
+            save_storage_policy(  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
                 config_manager,
                 selected_root=normalized_selected_root,
                 selection_source=payload.selection_source,
@@ -1746,7 +1791,10 @@ async def _post_storage_location_restart_locked(
             await _request_app_shutdown(request_app_shutdown)
         except Exception as exc:
             try:
-                _restore_storage_mutation_state(config_manager, state_snapshot, anchor_root=anchor_root)
+                # 与 _release_storage_startup_barrier_or_rollback 里那处同因同治：
+                # 这个 helper 末步 save_root_state，而无锁的 GET 路由也在事件循环上
+                # 读改写同一份 root_state，两者今天只靠「同在循环线程」互斥。详见那处注释。
+                _restore_storage_mutation_state(config_manager, state_snapshot, anchor_root=anchor_root)  # noqa: ASYNC_BLOCK — 同上：root_state 读改写的互斥依赖「同在循环线程」
             except Exception:
                 logger.exception(
                     "failed to rollback storage mutation state after restart scheduling failed",
@@ -1813,7 +1861,7 @@ async def _post_storage_location_restart_locked(
     except Exception as exc:
         try:
             if isinstance(previous_migration_payload, dict):
-                save_storage_migration(config_manager, previous_migration_payload, anchor_root=anchor_root)
+                save_storage_migration(config_manager, previous_migration_payload, anchor_root=anchor_root)  # noqa: ASYNC_BLOCK — 见 _STORAGE_MUTATION_STAYS_ON_LOOP
             else:
                 delete_storage_migration(config_manager, anchor_root=anchor_root)
         except Exception: