fix(lark): wake existing reply workers when durable work arrives

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -56,6 +56,19 @@ Lark queued notices describe the observed queue state without claiming that an
 earlier request is running: the first Turn can also await its native claim.
 Recovery retains an already persisted notice's exact text and provider intent.
 
+The existing four-slot reply pool wakes after private admission returns and
+after a local Turn worker exits, including preparation failure. These in-process
+Event hints carry no content or delivery authority and perform no provider IO;
+the scanner still reads the canonical journals and rechecks the source and Core.
+It clears the hint before discovery so arrivals during a scan remain visible to
+the following wait. Shutdown wakes that wait and unregisters the runtime hint.
+The one-second recovery poll, bounded pool, request fairness and thirty-second
+owner-notification cadence remain; custom controllers without the hook still poll.
+An occupied worker may still defer a request until a later scan. No completed-
+Future callback repeatedly reschedules unfinished work. File/Core and native
+protocol fixtures qualify this scheduling path separately from installed service,
+provider latency, complete stop acknowledgement and first-visible Web results.
+
 The packaged settings journey, source revocation/recovery, duplicate events,
 native queue/stop and two-App isolation have synthetic-provider regression and
 browser coverage. A native Codex source canary separately qualifies distinct

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -40,6 +40,15 @@ provider 读回确认回复；发生没有 receipt 的不确定写入时不盲
 飞书排队提示只表达已观测的队列状态，不断言已有前一条请求正在运行：首个 Turn
 也可能等待原生 claim。恢复继续使用已持久化提示的原文与 provider intent。
 
+私聊受理返回及本地 Turn worker 退出（含准备失败）时，唤醒既有四槽回复池。
+进程内 Event 只作提示，不携带内容或投递权限，也不执行 provider IO；扫描仍读取
+canonical journal，并重新核验来源与 Core。扫描前清除提示，避免扫描期间到达的
+信号在随后等待时丢失；关闭服务会唤醒等待并注销 runtime 提示。保留一秒恢复轮询、
+工作池上限、请求公平性和三十秒 owner 通知节奏；没有 hook 的自定义 controller
+继续轮询。工作槽被占用时，请求仍可能延至后续扫描；不通过已完成 Future 的回调
+反复重排尚未完成的工作。File/Core 与原生协议 fixture 单独验证这一调度路径，
+不替代安装态服务、provider 延迟、完整停止回执及 Web 首次可见结果的验收。
+
 源码 canary 已验证打包页面的桌面/窄屏、撤权拒绝与恢复、慢会话期间另一 App 完成、
 后续消息持久排队、exact stop，以及重启后原会话恢复和已确认回复不重复发送。
 这些是合成 provider/协议验收；真实原生 Codex 另行验证独立线程与上下文隔离。

loopx/chat_runtime.py

@@ -44,6 +44,7 @@
 from .chat_dsh import DshChatAdapter
 from .chat_endpoint_catalog import builtin_chat_endpoints
 from .chat_endpoints import AgentEndpointRegistry
+from .chat_turn_release import TurnReleaseWakeup
 from .control_plane.turn_driver.host_binding import MANAGED_TURN_HOST
 from .control_plane.turn_driver.execution_profile import managed_execution_profile
 from .kiro_cli_goal_mode import (
@@ -347,6 +348,7 @@ def __init__(
         self.cancelled_turns: set[tuple[str, str]] = set()
         self.turn_event_buffers: dict[tuple[str, str], _TurnEventBuffer] = {}
         self.turn_done_events: dict[tuple[str, str], threading.Event] = {}
+        self.turn_release_wakeup = TurnReleaseWakeup()
         self.lock = threading.RLock()
         self.session_open_locks: dict[tuple[str, str, str], threading.Lock] = {}
         self.session_adapter_locks: dict[str, threading.Lock] = {}
@@ -1197,6 +1199,7 @@ def _start_accepted_turn_worker(
                 if self.turn_done_events.get(key) is done_event:
                     self.turn_done_events.pop(key, None)
             done_event.set()
+            self.turn_release_wakeup.notify()
             raise
         return True
 
@@ -1541,6 +1544,7 @@ def _fail_queue_preparation(
             )
         finally:
             done_event.set()
+            self.turn_release_wakeup.notify()
             if owns_done_event:
                 with self.lock:
                     if self.turn_done_events.get(key) is done_event:
@@ -1596,6 +1600,7 @@ def _run_turn(
                     pass
         finally:
             done_event.set()
+            self.turn_release_wakeup.notify()
             with self.lock:
                 if self.turn_done_events.get(key) is done_event:
                     self.turn_done_events.pop(key, None)

loopx/chat_turn_release.py

@@ -0,0 +1,30 @@
+"""Content-free hints for existing local readers of the durable Turn journal."""
+from collections.abc import Iterator
+from contextlib import contextmanager
+import threading
+
+
+class TurnReleaseWakeup:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._events: set[threading.Event] = set()
+
+    @contextmanager
+    def __call__(self, event: threading.Event) -> Iterator[None]:
+        with self._lock:
+            if event in self._events:
+                raise ValueError("turn release wakeup is already registered")
+            self._events.add(event)
+        try:
+            yield
+        finally:
+            with self._lock:
+                self._events.discard(event)
+
+    def notify(self) -> None:
+        # Events do no IO and carry no Session, text or completion authority.
+        # Never invoke transport code from a model worker or under its lock.
+        with self._lock:
+            events = tuple(self._events)
+        for event in events:
+            event.set()

loopx/extensions/lark/goal_topic_runtime_service.py

@@ -4,8 +4,8 @@
 
 from collections.abc import Callable, Mapping
 from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import nullcontext
 from datetime import datetime, timezone
-from functools import partial
 import hashlib
 import logging
 from pathlib import Path
@@ -56,6 +56,7 @@ def __init__(
         self._workers: dict[str, ProfileWorker] = {}
         self._health: dict[str, dict[str, Any]] = {}
         self._closed = threading.Event()
+        self._delivery_wakeup = threading.Event()
         self._startup_thread: threading.Thread | None = None
         self._delivery_thread: threading.Thread | None = None
 
@@ -86,8 +87,15 @@ def _reconcile_private(self) -> None:
         attempted: dict[Path, int] = {}
         attention: Future[int] | None = None
         attention_due = 0.0
-        with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool:
+        wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None)
+        # Older/custom controllers retain polling. One local hint wakes the
+        # existing scanner; it is neither a queue nor a delivery receipt.
+        with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \
+                (wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()):
             while not self._closed.is_set():
+                # Clear BEFORE scanning: an arrival during the scan must also
+                # wake the following wait, rather than disappear in that gap.
+                self._delivery_wakeup.clear()
                 if attention is not None and attention.done():
                     try:
                         attention.result()
@@ -120,7 +128,17 @@ def _reconcile_private(self) -> None:
                         attention = pool.submit(owner_attention.reconcile)
                 except Exception:
                     logging.getLogger(__name__).warning("Private Chat delivery discovery is pending")
-                self._closed.wait(1)
+                self._delivery_wakeup.wait(1)
+
+    def _admit_private(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
+        if self._closed.is_set():
+            raise RuntimeError("Lark conversation service is closed")
+        result = self.private_conversations.admit(profile, event, defer_background_stop=True)
+        # Admission owns persistence and authorization. Only after it returns
+        # may the existing reply worker look for its durable work; duplicates
+        # and denied inputs still cannot manufacture a pending journal entry.
+        self._delivery_wakeup.set()
+        return result
 
     def _refresh_on_start(self) -> None:
         while not self._closed.is_set():
@@ -312,7 +330,7 @@ def handle_review_callback(
                             stop=stop,
                             runtime_root=self.runtime_root,
                             answer=answer,
-                            private_admitter=(partial(self.private_conversations.admit, defer_background_stop=True)
+                            private_admitter=(self._admit_private
                                 if self.private_conversations is not None else None),
                             proposal_deliverer=deliver_proposals,
                             review_callback_handler=(
@@ -486,6 +504,7 @@ def active_profiles(self) -> list[str]:
 
     def close(self) -> None:
         self._closed.set()
+        self._delivery_wakeup.set()
         with self._lock:
             workers = list(self._workers.values())
             self._workers.clear()