|
4 | 4 |
|
5 | 5 | from collections.abc import Callable, Mapping |
6 | 6 | from concurrent.futures import Future, ThreadPoolExecutor |
| 7 | +from contextlib import nullcontext |
7 | 8 | from datetime import datetime, timezone |
8 | | -from functools import partial |
9 | 9 | import hashlib |
10 | 10 | import logging |
11 | 11 | from pathlib import Path |
@@ -56,6 +56,7 @@ def __init__( |
56 | 56 | self._workers: dict[str, ProfileWorker] = {} |
57 | 57 | self._health: dict[str, dict[str, Any]] = {} |
58 | 58 | self._closed = threading.Event() |
| 59 | + self._delivery_wakeup = threading.Event() |
59 | 60 | self._startup_thread: threading.Thread | None = None |
60 | 61 | self._delivery_thread: threading.Thread | None = None |
61 | 62 |
|
@@ -86,8 +87,15 @@ def _reconcile_private(self) -> None: |
86 | 87 | attempted: dict[Path, int] = {} |
87 | 88 | attention: Future[int] | None = None |
88 | 89 | 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()): |
90 | 95 | 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() |
91 | 99 | if attention is not None and attention.done(): |
92 | 100 | try: |
93 | 101 | attention.result() |
@@ -120,7 +128,17 @@ def _reconcile_private(self) -> None: |
120 | 128 | attention = pool.submit(owner_attention.reconcile) |
121 | 129 | except Exception: |
122 | 130 | 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 |
124 | 142 |
|
125 | 143 | def _refresh_on_start(self) -> None: |
126 | 144 | while not self._closed.is_set(): |
@@ -312,7 +330,7 @@ def handle_review_callback( |
312 | 330 | stop=stop, |
313 | 331 | runtime_root=self.runtime_root, |
314 | 332 | answer=answer, |
315 | | - private_admitter=(partial(self.private_conversations.admit, defer_background_stop=True) |
| 333 | + private_admitter=(self._admit_private |
316 | 334 | if self.private_conversations is not None else None), |
317 | 335 | proposal_deliverer=deliver_proposals, |
318 | 336 | review_callback_handler=( |
@@ -486,6 +504,7 @@ def active_profiles(self) -> list[str]: |
486 | 504 |
|
487 | 505 | def close(self) -> None: |
488 | 506 | self._closed.set() |
| 507 | + self._delivery_wakeup.set() |
489 | 508 | with self._lock: |
490 | 509 | workers = list(self._workers.values()) |
491 | 510 | self._workers.clear() |
|
0 commit comments