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

app/main_server/workshop_runtime.py

@@ -26,7 +26,7 @@
     warmup_ugc_cache,
 )
 from utils.cloudsave_runtime import is_write_fence_active
-from utils.workshop_utils import get_workshop_path, get_workshop_root
+from utils.workshop_utils import get_workshop_path_async, get_workshop_root_async
 
 from ._shared import runtime
 
@@ -127,7 +127,7 @@ async def _init_and_mount_workshop():
             subscribed_items = workshop_items_result.get("items", [])
 
         # 3. 调用 utils 层函数获取/计算路径（路径会被持久化到 config）
-        workshop_path = get_workshop_root(subscribed_items)
+        workshop_path = await get_workshop_root_async(subscribed_items)
 
         # 4. 挂载静态文件目录
         if (
@@ -149,7 +149,7 @@ async def _init_and_mount_workshop():
     except Exception as e:
         logger.error(f"初始化创意工坊目录时出错: {e}")
         # 降级：确保至少有一个默认路径可用
-        workshop_path = get_workshop_path()
+        workshop_path = await get_workshop_path_async()
         logger.info(f"使用配置中的默认路径: {workshop_path}")
         if (
             workshop_path

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

docs/api/rest/workshop.md

@@ -77,6 +77,10 @@ Synchronization may report skipped/conflicting cards, missing installs, or a sto
 
 These routes package reference material; they do not clone or register a local TTS voice themselves.
 
+::: info Content-folder exclusivity
+Publishing hands the whole content folder to Steam until the upload finishes. While a folder is publishing, `upload-reference-audio`, `remove-reference-audio` and `cleanup-temp-folder` answer `409` instead of modifying bytes that Steam is consuming. The exclusion also works in reverse: `publish` answers `409` while a reference-audio write is in flight.
+:::
+
 ## Implementation-verified route inventory
 
 ```text

docs/ja/api/rest/workshop.md

@@ -77,6 +77,10 @@ Sync は skip/conflict、missing install、storage write fence を JSON で報
 
 Reference material を package するだけで、local TTS voice の clone/register は行いません。
 
+::: info Content folder の排他
+publish は upload 完了まで content folder 全体を Steam に渡します。その間は `upload-reference-audio`、`remove-reference-audio`、`cleanup-temp-folder` が待たずに `409` を返し、Steam が使用中の bytes を変更しません。逆方向も同じで、reference audio の書き込み中は `publish` が `409` を返します。
+:::
+
 ## 実装で確認した route 一覧
 
 ```text

docs/zh-CN/api/rest/workshop.md

@@ -77,6 +77,10 @@ macOS arm64 上当前 Steamworks 绑定存在 callback 崩溃风险，因此原
 
 这些接口只打包参考材料，不会自行克隆或注册本地 TTS 声音。
 
+::: info 内容目录互斥
+发布会在上传结束前把整个内容目录交给 Steam。一个目录正在发布时，`upload-reference-audio`、`remove-reference-audio` 和 `cleanup-temp-folder` 直接返回 `409`，不会改动 Steam 正在读取的内容；反过来也一样，参考声音还在写入时，`publish` 返回 `409`。
+:::
+
 ## 经实现核对的路由清单
 
 ```text

main_logic/core/notify.py

@@ -203,7 +203,9 @@ async def _build_initial_prompt(self) -> str:
         try:
             from memory.anti_repeat import get_anti_repeat_corpus
             from config.prompts.prompts_directives import render_recent_topics_block
-            topics = get_anti_repeat_corpus().top_recent_topics(_directives_key)
+            anti_repeat_corpus = get_anti_repeat_corpus()
+            await anti_repeat_corpus.apreload(_directives_key)
+            topics = anti_repeat_corpus.top_recent_topics(_directives_key)
             prompt += render_recent_topics_block(topics, _lang)
         except Exception as _exc:  # pragma: no cover - defensive
             logger.debug(

main_logic/core/proactive.py

@@ -260,6 +260,14 @@ def _user_active_recently() -> bool:
             if self.state.is_proactive_preempted():
                 logger.info("[%s] prepare_proactive_delivery: preempted during auto-start", self.lanlan_name)
                 return False
+        # ``finish_proactive_delivery`` must stage the committed text before
+        # terminal signals and cannot insert an await there. Pay the first
+        # corpus read here, before the proactive turn is claimed or visible.
+        try:
+            from memory.anti_repeat import get_anti_repeat_corpus
+            await get_anti_repeat_corpus().apreload(self.lanlan_name)
+        except Exception as exc:  # pragma: no cover - best-effort cache
+            logger.debug("[AntiRepeat] proactive preload skipped: %s", exc)
         async with self.lock:
             # lock 内二次复查：USER_INPUT 在 self.lock 内 rotate sid，sticky preempt
             # flag 先于 sid mutation 翻起；此处若已被抢占则不写 current_speech_id。
@@ -483,26 +491,32 @@ async def finish_proactive_delivery(
                 # 绝不会为未投递的轮次暂存截图。
                 if hasattr(self.session, "set_proactive_screenshot"):
                     self.session.set_proactive_screenshot(vision_screenshot_b64)
+
+            # 防复读 corpus 拆成两半：内存更新在收尾信号**之前**（同步、无 await，
+            # 所以不是取消点），落盘在之后。用户可能对着主动搭话立刻回一句，那一轮
+            # 打分必须已经看得到刚投递的这段；而落盘那个 await 一旦被取消就会跳过
+            # TTS 收尾和两处 turn end。两个要求方向相反，只有拆开才能同时满足。
+            #
+            # LLM 给自己的元数据备忘，不算复读对象。素材推送类 channel（推歌）
+            # 的台词天生模板化，录进 corpus 会污染 FG 窗、漂移其它 channel 的
+            # 复读基线，故按 ANTI_REPEAT_EXEMPT_SOURCE_TAGS 豁免（与出口的
+            # BM25 评分豁免对偶）。
+            staged_anti_repeat = None
+            if source_tag not in ANTI_REPEAT_EXEMPT_SOURCE_TAGS:
+                try:
+                    from memory.anti_repeat import get_anti_repeat_corpus
+                    staged_anti_repeat = get_anti_repeat_corpus().stage_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] stage proactive skipped: %s", _exc)
                 # 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
-                        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 +533,19 @@ async def finish_proactive_delivery(
             except Exception:
                 # Turn-end push is best-effort; the client may have gone away.
                 pass
+
+            # 落盘排在所有收尾信号之后（内存更新已经在投递后立刻做了，见上），而且
+            # **摘下来不 await**：到这里这一轮对用户已经发生完了，但下面那句
+            # `return True` 才是调用方的记账凭据（break reminder / 小游戏邀请看它决定
+            # 要不要把这条来源标记成已消费）。在这里 await 就等于把一个取消点插在
+            # 「已投递」和「报告已投递」之间 —— CancelledError 是 BaseException，
+            # 下面的 except Exception 接不住，同一条提醒会被再发一次。
+            if staged_anti_repeat is not None:
+                try:
+                    from memory.anti_repeat import get_anti_repeat_corpus
+                    get_anti_repeat_corpus().flush_staged_detached(staged_anti_repeat)
+                except Exception as _exc:  # pragma: no cover
+                    logger.debug("[AntiRepeat] flush 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

@@ -183,6 +183,16 @@ async def prompt_ephemeral(
         if not instruction or not instruction.strip():
             return False
 
+        # A regular visible response stages anti-repeat memory immediately
+        # before terminal callbacks. That commit boundary cannot gain an await,
+        # so perform the first disk-backed load before generation starts.
+        if completion_mode == "response" and persist_response:
+            try:
+                from memory.anti_repeat import get_anti_repeat_corpus
+                await get_anti_repeat_corpus().apreload(self.lanlan_name)
+            except Exception as exc:  # pragma: no cover - best-effort cache
+                logger.debug("[AntiRepeat] response preload skipped: %s", exc)
+
         # 临时注入：instruction 已由调用方用 ======== 格式封装，作为 HumanMessage 发送，
         # 不持久化到 _conversation_history，避免污染长期上下文。
         # Proactive media is passed EXPLICITLY via ``images`` (per-callback,
@@ -432,22 +442,36 @@ 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))
-                # 防复读 corpus：只录常规 reply（completion_mode == "response"）。
-                # proactive 路径已经在 ``core.finish_proactive_delivery`` 上录，
-                # 这里再录会双写——这两条路径都接得到同一段 assistant 文本。
-                if completion_mode == "response":
+            # 防复读 corpus 拆成两半：内存更新在收尾信号**之前**（同步，不含 await，
+            # 所以不是取消点），落盘在**之后**。客户端看到 turn end 就可能立刻发下一
+            # 条，那一轮的打分必须已经看得到刚提交的这句；而落盘那个 await 一旦被取消
+            # 就会跳过 on_response_done 里的 TTS 收尾 / turn 结束 / request-id 清理。
+            # 两个要求方向相反，只有拆开才能同时满足。
+            staged_anti_repeat = None
+            if completion_mode == "response" and content_committed and persist_response:
+                try:
+                    from memory.anti_repeat import get_anti_repeat_corpus
+                    staged_anti_repeat = get_anti_repeat_corpus().stage_output(
+                        self.lanlan_name, committed_text, is_proactive=False,
+                    )
+                except Exception as _exc:  # pragma: no cover
+                    logger.debug("[AntiRepeat] stage reply skipped: %s", _exc)
+            if completion_mode == "response":
+                if self.on_response_done:
+                    await self.on_response_done()
+                # 只录常规 reply（completion_mode == "response"）。proactive 路径
+                # 已经在 ``core.finish_proactive_delivery`` 上录，这里再录会双写。
+                # 与 core.finish_proactive_delivery 同因同治：摘下来不 await。下面的
+                # `return content_committed` 是调用方判断这轮有没有提交的依据，在它
+                # 之前留一个取消点，就会让一次已经发出去的回复被记成没发。
+                if staged_anti_repeat is not None:
                     try:
                         from memory.anti_repeat import get_anti_repeat_corpus
-                        get_anti_repeat_corpus().record_output(
-                            self.lanlan_name, committed_text, is_proactive=False,
-                        )
+                        get_anti_repeat_corpus().flush_staged_detached(staged_anti_repeat)
                     except Exception as _exc:  # pragma: no cover
                         logger.debug(
-                            "[AntiRepeat] record reply skipped: %s", _exc,
+                            "[AntiRepeat] flush 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_logic/proactive_chat/break_reminders.py

@@ -309,6 +309,11 @@ async def _deliver_break_reminder_via_llm(
     if not await mgr.prepare_proactive_delivery(min_idle_secs=10.0):
         return BreakReminderDeliveryResult()
 
+    try:
+        await get_anti_repeat_corpus().apreload(lanlan_name)
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("[AntiRepeat] break-reminder preload skipped: %s", exc)
+
     silence_since_before_generation = _break_reminder_silence_since(mgr)
     proactive_sid = mgr.current_speech_id
     from main_logic.session_state import SessionEvent as _SE

main_logic/proactive_chat/generation.py

@@ -1180,6 +1180,7 @@ def _output(
         from memory.anti_repeat import get_anti_repeat_corpus
 
         anti_repeat_corpus = get_anti_repeat_corpus()
+        await anti_repeat_corpus.apreload(lanlan_name)
     except Exception as exc:  # pragma: no cover - defensive
         active_logger.debug("[AntiRepeat] corpus unavailable: %s", exc)
         anti_repeat_corpus = None