From dc3feb5ce85a0b4bd8879b8b7be465bbd6cdf5b6 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 11:29:10 +0800 Subject: [PATCH] fix(lark): wake existing delivery workers on durable work Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 13 + ...p-conversation-and-async-inbox-v0.zh-CN.md | 9 + loopx/chat_runtime.py | 5 + loopx/chat_turn_release.py | 30 +++ .../lark/goal_topic_runtime_service.py | 27 ++- tests/test_lark_delivery_wakeup.py | 227 ++++++++++++++++++ 6 files changed, 307 insertions(+), 4 deletions(-) create mode 100644 loopx/chat_turn_release.py create mode 100644 tests/test_lark_delivery_wakeup.py diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 4316bb9734..ccdfd9b6ca 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 695efba422..b2429ee4e3 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/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 另行验证独立线程与上下文隔离。 diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 347b0a6e7d..6a8d8c8d42 100644 --- a/loopx/chat_runtime.py +++ b/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) diff --git a/loopx/chat_turn_release.py b/loopx/chat_turn_release.py new file mode 100644 index 0000000000..eadccc2c15 --- /dev/null +++ b/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() diff --git a/loopx/extensions/lark/goal_topic_runtime_service.py b/loopx/extensions/lark/goal_topic_runtime_service.py index 060e5e9258..0ca7218481 100644 --- a/loopx/extensions/lark/goal_topic_runtime_service.py +++ b/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() diff --git a/tests/test_lark_delivery_wakeup.py b/tests/test_lark_delivery_wakeup.py new file mode 100644 index 0000000000..4f50e029e5 --- /dev/null +++ b/tests/test_lark_delivery_wakeup.py @@ -0,0 +1,227 @@ +"""Durable File/Core journeys with polling held behind an explicit barrier.""" +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +import json +from pathlib import Path +import threading +from types import SimpleNamespace + +import pytest + +from test_chat_ordinary_project import ordinary # noqa: F401 +from test_lark_private_conversations import connect +from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService + + +@contextmanager +def held_poll(service, monkeypatch): + """A hint can wake work, but elapsed time cannot rescue the assertion.""" + idle, release_fallback = threading.Event(), threading.Event() + original_closed_wait = service._closed.wait + + def old_poll(_timeout): + idle.set() + assert release_fallback.wait(10) + return service._closed.is_set() + + monkeypatch.setattr(service._closed, "wait", old_poll) + wakeup = getattr(service, "_delivery_wakeup", None) + if wakeup is not None: + original_work_wait = wakeup.wait + + def hinted_poll(_timeout): + idle.set() + return original_work_wait(10) + + monkeypatch.setattr(wakeup, "wait", hinted_poll) + worker = threading.Thread(target=service._reconcile_private) + worker.start() + try: + yield idle + finally: + service.close() + release_fallback.set() + if wakeup is not None: + wakeup.set() + worker.join(10) + assert not worker.is_alive() + monkeypatch.setattr(service._closed, "wait", original_closed_wait) + + +@pytest.mark.parametrize("arrival", ["during_scan", "during_wait"]) +def test_existing_listener_wakes_original_journal_without_polling(ordinary, monkeypatch, tmp_path, arrival): # noqa: F811 + import loopx.extensions.lark.goal_topic_runtime as provider_runtime + + store, runtime, provider, transport = connect(ordinary) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=store.root.parent, + runtime_controller=runtime, private_conversations=transport) + event = provider.event("notes-app", "wake-status", "/status") + scan_entered, admitted, feed = threading.Event(), threading.Event(), threading.Event() + returned, sent = threading.Event(), threading.Event() + original_paths, original_reconcile = transport.pending_delivery_paths, transport.reconcile_request + scans = 0 + + def paths(): + nonlocal scans + snapshot = original_paths() + scans += 1 + if arrival == "during_scan" and scans == 1: + scan_entered.set() + assert admitted.wait(5) + return snapshot + + def reconcile(path): + result = original_reconcile(path) + if result: + sent.set() + return result + + active = True + + def stream(**kwargs): + nonlocal active + assert feed.wait(5) + try: + result = kwargs["private_admitter"]("notes-app", event) + assert result["status"] == "command_recorded" + admitted.set() + return {"status": "configuration_removed"} + finally: + active = False + kwargs["stop"].set() + returned.set() + + monkeypatch.setattr(transport, "pending_delivery_paths", paths) + monkeypatch.setattr(transport, "reconcile_request", reconcile) + monkeypatch.setattr(provider_runtime, "_active_profile_configs", lambda _: + {"notes-app": {"cli_bin": "lark-cli", "bot_app_id": "cli_notes_app"}} if active else {}) + monkeypatch.setattr(provider_runtime, "stream_lark_goal_topic_profile", stream) + # The real App consumer lease is exercised in this fixture's own home. + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + listener = threading.Thread(target=service._poll_profile, args=("notes-app", threading.Event())) + try: + with held_poll(service, monkeypatch) as idle: + assert (scan_entered if arrival == "during_scan" else idle).wait(5) + listener.start() + feed.set() + assert admitted.wait(5) and returned.wait(5) + # The timeout branch remains held. Only admission's wakeup can + # make this original request reach the existing reply worker. + assert sent.wait(5) + assert len(provider.writes) == 1 and provider.writes[0][0] == "notes-app" + assert "个人助手 · notes" in provider.writes[0][1] + assert "排队消息：0 条" in provider.writes[0][1] + assert "只读授权" in provider.writes[0][1] + assert store.list_sessions() == [] + rows = transport.core.pending() + assert len(rows) == 1 and rows[0]["status"] == "command_completed" + before = list(provider.writes) + assert transport.reconcile() == 0 and provider.writes == before + finally: + feed.set() + admitted.set() + if listener.ident is not None: + listener.join(10) + assert not listener.is_alive() + runtime.close() + assert not runtime.turn_release_wakeup._events + + +def test_native_terminal_release_wakes_final_after_prior_pending_read(ordinary, monkeypatch, tmp_path): # noqa: F811 + import loopx.extensions.lark.goal_topic_runtime_service as service_module + + store, runtime, provider, transport = connect(ordinary) + gate = tmp_path / "allow-native-completion" + fake = ordinary[-2] + original = " response = 'Visible runtime response." + replacement = (f" while not __import__('pathlib').Path({str(gate)!r}).exists():\n" + " time.sleep(0.01)\n" + original) + assert original in fake.read_text() + fake.write_text(fake.read_text().replace(original, replacement, 1)) + event = provider.event("notes-app", "native-wakeup", "Explain this project") + assert transport.admit("notes-app", event)["status"] == "durably_accepted" + native = transport.core.pending()[0] + pending_read_done, delivered = threading.Event(), threading.Event() + original_reconcile = transport.reconcile_request + + def reconcile(path): + result = original_reconcile(path) + if result: + delivered.set() + return result + + class ObservedPool(ThreadPoolExecutor): + def submit(self, fn, /, *args, **kwargs): + future = super().submit(fn, *args, **kwargs) + + def finished(done): + if done.exception() is None and done.result() == 0: + pending_read_done.set() + + future.add_done_callback(finished) + return future + + monkeypatch.setattr(transport, "reconcile_request", reconcile) + monkeypatch.setattr(service_module, "ThreadPoolExecutor", ObservedPool) + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=store.root.parent, + runtime_controller=runtime, private_conversations=transport) + try: + with held_poll(service, monkeypatch) as idle: + assert idle.wait(5) and pending_read_done.wait(5) + assert not delivered.is_set() + gate.touch() + turn = runtime.wait_for_turn(session_id=native["session_id"], turn_id=native["turn_id"], timeout_sec=5) + assert turn["status"] == "completed" + assert delivered.wait(5) + assert provider.writes[-1] == ("notes-app", turn["response"]["message"]) + assert provider.writes.count(provider.writes[-1]) == 1 + assert len(store.list_sessions()) == 1 + packets = [json.loads(line) for line in ordinary[4].read_text().splitlines()] + assert sum(p["method"] == "turn/start" for p in packets) == 1 + finally: + gate.touch() + runtime.close() + assert not runtime.turn_release_wakeup._events + + +def test_preparation_failure_notifies_reader_but_keeps_failed_turn(ordinary): # noqa: F811 + store, runtime, contexts, request, *_ = ordinary + _, opened = request("/api/chat/sessions", {"context_kind": "project", + "project_ref": contexts.available()[0]["project_ref"]}) + sid = opened["session_id"] + queued, _ = store.create_queued_turn(sid, client_turn_id="owned-prepare-failure", message="Explain") + wakeup = threading.Event() + try: + with runtime.turn_release_wakeup(wakeup): + with pytest.raises(ValueError, match="already registered"): + with runtime.turn_release_wakeup(wakeup): + pytest.fail("one reader must not register the same wakeup twice") + runtime._fail_queue_preparation(sid, queued["turn_id"], OSError("fixture runtime unavailable")) + assert wakeup.is_set() + assert store.load_turn(sid, queued["turn_id"])["status"] == "failed" + wakeup.clear() + runtime.turn_release_wakeup.notify() + assert not wakeup.is_set() and not runtime.turn_release_wakeup._events + finally: + runtime.close() + + +def test_closed_service_rejects_new_admission_without_dispatch(tmp_path): + def unexpected(*_args, **_kwargs): + pytest.fail("closed service must not admit a new input") + + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=None, private_conversations=SimpleNamespace(admit=unexpected)) + service.close() + with pytest.raises(RuntimeError, match="service is closed"): + service._admit_private("notes-app", {}) + + +def test_no_private_transport_registers_no_runtime_wakeup(tmp_path): + def unexpected(*_args): + pytest.fail("disabled transport must not register or start a worker") + + service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=tmp_path, + runtime_controller=SimpleNamespace(turn_release_wakeup=unexpected)) + service._reconcile_private() + service.close()