Skip to content

Commit cd14893

Browse files
authored
Merge pull request #6703 from loopx-project/codex/lark-delivery-wakeup-20261017
fix(lark): wake existing reply workers when durable work arrives
2 parents fdf03ec + dc3feb5 commit cd14893

6 files changed

Lines changed: 307 additions & 4 deletions

File tree

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,19 @@ Lark queued notices describe the observed queue state without claiming that an
5656
earlier request is running: the first Turn can also await its native claim.
5757
Recovery retains an already persisted notice's exact text and provider intent.
5858

59+
The existing four-slot reply pool wakes after private admission returns and
60+
after a local Turn worker exits, including preparation failure. These in-process
61+
Event hints carry no content or delivery authority and perform no provider IO;
62+
the scanner still reads the canonical journals and rechecks the source and Core.
63+
It clears the hint before discovery so arrivals during a scan remain visible to
64+
the following wait. Shutdown wakes that wait and unregisters the runtime hint.
65+
The one-second recovery poll, bounded pool, request fairness and thirty-second
66+
owner-notification cadence remain; custom controllers without the hook still poll.
67+
An occupied worker may still defer a request until a later scan. No completed-
68+
Future callback repeatedly reschedules unfinished work. File/Core and native
69+
protocol fixtures qualify this scheduling path separately from installed service,
70+
provider latency, complete stop acknowledgement and first-visible Web results.
71+
5972
The packaged settings journey, source revocation/recovery, duplicate events,
6073
native queue/stop and two-App isolation have synthetic-provider regression and
6174
browser coverage. A native Codex source canary separately qualifies distinct

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,15 @@ provider 读回确认回复;发生没有 receipt 的不确定写入时不盲
4040
飞书排队提示只表达已观测的队列状态,不断言已有前一条请求正在运行:首个 Turn
4141
也可能等待原生 claim。恢复继续使用已持久化提示的原文与 provider intent。
4242

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

‎loopx/chat_runtime.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
from .chat_dsh import DshChatAdapter
4545
from .chat_endpoint_catalog import builtin_chat_endpoints
4646
from .chat_endpoints import AgentEndpointRegistry
47+
from .chat_turn_release import TurnReleaseWakeup
4748
from .control_plane.turn_driver.host_binding import MANAGED_TURN_HOST
4849
from .control_plane.turn_driver.execution_profile import managed_execution_profile
4950
from .kiro_cli_goal_mode import (
@@ -347,6 +348,7 @@ def __init__(
347348
self.cancelled_turns: set[tuple[str, str]] = set()
348349
self.turn_event_buffers: dict[tuple[str, str], _TurnEventBuffer] = {}
349350
self.turn_done_events: dict[tuple[str, str], threading.Event] = {}
351+
self.turn_release_wakeup = TurnReleaseWakeup()
350352
self.lock = threading.RLock()
351353
self.session_open_locks: dict[tuple[str, str, str], threading.Lock] = {}
352354
self.session_adapter_locks: dict[str, threading.Lock] = {}
@@ -1197,6 +1199,7 @@ def _start_accepted_turn_worker(
11971199
if self.turn_done_events.get(key) is done_event:
11981200
self.turn_done_events.pop(key, None)
11991201
done_event.set()
1202+
self.turn_release_wakeup.notify()
12001203
raise
12011204
return True
12021205

@@ -1541,6 +1544,7 @@ def _fail_queue_preparation(
15411544
)
15421545
finally:
15431546
done_event.set()
1547+
self.turn_release_wakeup.notify()
15441548
if owns_done_event:
15451549
with self.lock:
15461550
if self.turn_done_events.get(key) is done_event:
@@ -1596,6 +1600,7 @@ def _run_turn(
15961600
pass
15971601
finally:
15981602
done_event.set()
1603+
self.turn_release_wakeup.notify()
15991604
with self.lock:
16001605
if self.turn_done_events.get(key) is done_event:
16011606
self.turn_done_events.pop(key, None)

‎loopx/chat_turn_release.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
"""Content-free hints for existing local readers of the durable Turn journal."""
2+
from collections.abc import Iterator
3+
from contextlib import contextmanager
4+
import threading
5+
6+
7+
class TurnReleaseWakeup:
8+
def __init__(self) -> None:
9+
self._lock = threading.Lock()
10+
self._events: set[threading.Event] = set()
11+
12+
@contextmanager
13+
def __call__(self, event: threading.Event) -> Iterator[None]:
14+
with self._lock:
15+
if event in self._events:
16+
raise ValueError("turn release wakeup is already registered")
17+
self._events.add(event)
18+
try:
19+
yield
20+
finally:
21+
with self._lock:
22+
self._events.discard(event)
23+
24+
def notify(self) -> None:
25+
# Events do no IO and carry no Session, text or completion authority.
26+
# Never invoke transport code from a model worker or under its lock.
27+
with self._lock:
28+
events = tuple(self._events)
29+
for event in events:
30+
event.set()

‎loopx/extensions/lark/goal_topic_runtime_service.py‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,8 @@
44

55
from collections.abc import Callable, Mapping
66
from concurrent.futures import Future, ThreadPoolExecutor
7+
from contextlib import nullcontext
78
from datetime import datetime, timezone
8-
from functools import partial
99
import hashlib
1010
import logging
1111
from pathlib import Path
@@ -56,6 +56,7 @@ def __init__(
5656
self._workers: dict[str, ProfileWorker] = {}
5757
self._health: dict[str, dict[str, Any]] = {}
5858
self._closed = threading.Event()
59+
self._delivery_wakeup = threading.Event()
5960
self._startup_thread: threading.Thread | None = None
6061
self._delivery_thread: threading.Thread | None = None
6162

@@ -86,8 +87,15 @@ def _reconcile_private(self) -> None:
8687
attempted: dict[Path, int] = {}
8788
attention: Future[int] | None = None
8889
attention_due = 0.0
89-
with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool:
90+
wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None)
91+
# Older/custom controllers retain polling. One local hint wakes the
92+
# existing scanner; it is neither a queue nor a delivery receipt.
93+
with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \
94+
(wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()):
9095
while not self._closed.is_set():
96+
# Clear BEFORE scanning: an arrival during the scan must also
97+
# wake the following wait, rather than disappear in that gap.
98+
self._delivery_wakeup.clear()
9199
if attention is not None and attention.done():
92100
try:
93101
attention.result()
@@ -120,7 +128,17 @@ def _reconcile_private(self) -> None:
120128
attention = pool.submit(owner_attention.reconcile)
121129
except Exception:
122130
logging.getLogger(__name__).warning("Private Chat delivery discovery is pending")
123-
self._closed.wait(1)
131+
self._delivery_wakeup.wait(1)
132+
133+
def _admit_private(self, profile: str, event: dict[str, Any]) -> dict[str, Any]:
134+
if self._closed.is_set():
135+
raise RuntimeError("Lark conversation service is closed")
136+
result = self.private_conversations.admit(profile, event, defer_background_stop=True)
137+
# Admission owns persistence and authorization. Only after it returns
138+
# may the existing reply worker look for its durable work; duplicates
139+
# and denied inputs still cannot manufacture a pending journal entry.
140+
self._delivery_wakeup.set()
141+
return result
124142

125143
def _refresh_on_start(self) -> None:
126144
while not self._closed.is_set():
@@ -312,7 +330,7 @@ def handle_review_callback(
312330
stop=stop,
313331
runtime_root=self.runtime_root,
314332
answer=answer,
315-
private_admitter=(partial(self.private_conversations.admit, defer_background_stop=True)
333+
private_admitter=(self._admit_private
316334
if self.private_conversations is not None else None),
317335
proposal_deliverer=deliver_proposals,
318336
review_callback_handler=(
@@ -486,6 +504,7 @@ def active_profiles(self) -> list[str]:
486504

487505
def close(self) -> None:
488506
self._closed.set()
507+
self._delivery_wakeup.set()
489508
with self._lock:
490509
workers = list(self._workers.values())
491510
self._workers.clear()

0 commit comments

Comments
 (0)