From ab7ab449f6626348ea70a5f0900799272c833530 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=BDofka?= Date: Sun, 17 May 2026 10:37:55 +0200 Subject: [PATCH 1/4] fix(simplex): add polling fallback for missed chat items Reduce fallback polling latency, persist seen item IDs, and resolve SimpleX send targets via CLI display names. --- plugins/platforms/simplex/adapter.py | 330 ++++++++++++++++++++++++--- tests/gateway/test_simplex_plugin.py | 129 ++++++++++- 2 files changed, 427 insertions(+), 32 deletions(-) diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index b568f29bbb5e8..f7992b9e07b3a 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -38,6 +38,7 @@ import random import time from datetime import datetime, timezone +from pathlib import Path from typing import Any, Dict, List, Optional # Lazy import: BasePlatformAdapter and friends live in the main repo. @@ -53,6 +54,7 @@ cache_audio_from_bytes, cache_document_from_bytes, ) +from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) @@ -65,6 +67,12 @@ WS_RETRY_DELAY_MAX = 60.0 HEALTH_CHECK_INTERVAL = 30.0 HEALTH_CHECK_STALE_THRESHOLD = 120.0 +POLL_INTERVAL = 1.0 +# Polling is the reliability path when daemon push events are flaky. Keep its +# command timeout short: a missed `/tail 50` should cost one beat, not turn into +# the 50–60s "SimpleX feels dead" delay after a few consecutive stalls. +POLL_COMMAND_TIMEOUT = 2.0 +POLL_CONNECT_TIMEOUT = 2.0 # Correlation ID prefix for requests we send so we can ignore our own echoes. _CORR_PREFIX = "hermes-" @@ -108,6 +116,11 @@ def _is_audio_ext(ext: str) -> bool: return ext.lower() in (".mp3", ".wav", ".ogg", ".m4a", ".aac") +def _simplex_quote_name(name: str) -> str: + """Quote a simplex-chat display name for command targets.""" + return str(name).replace("'", "\\'") + + # --------------------------------------------------------------------------- # SimpleX Adapter # --------------------------------------------------------------------------- @@ -130,9 +143,13 @@ def __init__(self, config: PlatformConfig, **kwargs): self._ws = None # websockets connection self._ws_task: Optional[asyncio.Task] = None self._health_task: Optional[asyncio.Task] = None + self._poll_task: Optional[asyncio.Task] = None self._typing_tasks: Dict[str, asyncio.Task] = {} self._running = False self._last_ws_activity = 0.0 + self._seen_item_ids: set = set() + self._seen_items_path = get_hermes_home() / "simplex_seen_items.json" + self._connected_at = 0.0 # Track sent correlation IDs to filter echoes self._pending_corr_ids: set = set() @@ -170,8 +187,12 @@ async def connect(self) -> bool: self._running = True self._last_ws_activity = time.time() + self._connected_at = time.time() + self._load_seen_items() + await self._seed_seen_items() self._ws_task = asyncio.create_task(self._ws_listener()) self._health_task = asyncio.create_task(self._health_monitor()) + self._poll_task = asyncio.create_task(self._poll_unread_items()) logger.info("SimpleX: connected to %s", self.ws_url) return True @@ -194,6 +215,13 @@ async def disconnect(self) -> None: except asyncio.CancelledError: pass + if self._poll_task: + self._poll_task.cancel() + try: + await self._poll_task + except asyncio.CancelledError: + pass + for task in self._typing_tasks.values(): task.cancel() self._typing_tasks.clear() @@ -207,6 +235,59 @@ async def disconnect(self) -> None: logger.info("SimpleX: disconnected") + def _item_key(self, wrapper: dict) -> Optional[str]: + """Return a stable per-chat key for a SimpleX chat item.""" + chat_info = wrapper.get("chatInfo") or wrapper.get("chat") or {} + chat_item = wrapper.get("chatItem") or wrapper.get("item") or {} + meta = chat_item.get("meta") or {} + item_id = meta.get("itemId") + if item_id is None: + return None + chat_type = chat_info.get("type") or "" + if chat_type in ("group", "groupInfo"): + group_info = chat_info.get("groupInfo") or chat_info.get("group") or {} + chat_id = group_info.get("groupId") or group_info.get("id") or "" + prefix = "group" + else: + contact_info = chat_info.get("contact") or {} + chat_id = contact_info.get("contactId") or contact_info.get("id") or "" + prefix = "direct" + if not chat_id: + return None + return f"{prefix}:{chat_id}:{item_id}" + + def _item_timestamp_epoch(self, wrapper: dict) -> Optional[float]: + chat_item = wrapper.get("chatItem") or wrapper.get("item") or {} + meta = chat_item.get("meta") or {} + ts_str = meta.get("itemTs") or meta.get("createdAt") or "" + if not ts_str: + return None + try: + return datetime.fromisoformat(str(ts_str).replace("Z", "+00:00")).timestamp() + except (ValueError, AttributeError): + return None + + def _load_seen_items(self) -> None: + try: + data = json.loads(Path(self._seen_items_path).read_text(encoding="utf-8")) + if isinstance(data, list): + self._seen_item_ids.update(str(x) for x in data if x) + except FileNotFoundError: + pass + except Exception as e: + logger.debug("SimpleX poll: failed to load seen items: %s", e) + + def _save_seen_items(self) -> None: + try: + items = list(self._seen_item_ids)[-1000:] + self._seen_item_ids = set(items) + Path(self._seen_items_path).write_text( + json.dumps(items, ensure_ascii=False), + encoding="utf-8", + ) + except Exception as e: + logger.debug("SimpleX poll: failed to save seen items: %s", e) + # ------------------------------------------------------------------ # WebSocket listener # ------------------------------------------------------------------ @@ -287,6 +368,157 @@ async def _health_monitor(self) -> None: except Exception: pass + # ------------------------------------------------------------------ + # Polling fallback + # ------------------------------------------------------------------ + + async def _command_once( + self, + cmd: str, + *, + timeout: float = 10.0, + open_timeout: float = 10.0, + ) -> Optional[dict]: + """Run one simplex-chat command over an ephemeral WebSocket. + + The daemon does not reliably push ``newChatItem`` events to every + persistent WebSocket client in all versions/modes. A tiny polling + fallback makes inbound DM/group delivery deterministic while keeping + the persistent listener for installations where push events work. + """ + import websockets as _wsclient + + corr_id = self._make_corr_id() + payload = {"corrId": corr_id, "cmd": cmd} + try: + async with _wsclient.connect(self.ws_url, open_timeout=open_timeout, close_timeout=1) as ws: + await ws.send(json.dumps(payload)) + deadline = time.time() + timeout + while time.time() < deadline: + raw = await asyncio.wait_for(ws.recv(), timeout=max(0.1, deadline - time.time())) + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + if msg.get("corrId") == corr_id: + self._pending_corr_ids.discard(corr_id) + return msg + finally: + self._pending_corr_ids.discard(corr_id) + return None + + async def _seed_seen_items(self) -> None: + """Remember current history so a gateway restart doesn't answer old mail.""" + try: + resp = await self._command_once("/tail 50") + changed = False + for wrapper in (resp or {}).get("resp", {}).get("chatItems", []) or []: + item_key = self._item_key(wrapper) + if item_key is not None: + self._seen_item_ids.add(item_key) + changed = True + if changed: + self._save_seen_items() + logger.info("SimpleX poll: seeded %d seen chat items", len(self._seen_item_ids)) + except Exception as e: + logger.debug("SimpleX poll: initial seed failed: %s", e) + + async def _poll_unread_items(self) -> None: + """Poll recent chat history for unread inbound items missed by push WS.""" + while self._running: + await asyncio.sleep(POLL_INTERVAL) + try: + start = time.time() + resp = await self._command_once( + "/tail 50", + timeout=POLL_COMMAND_TIMEOUT, + open_timeout=POLL_CONNECT_TIMEOUT, + ) + poll_elapsed = time.time() - start + wrappers = (resp or {}).get("resp", {}).get("chatItems", []) or [] + if resp is None: + logger.warning( + "SimpleX poll: /tail 50 timed out after %.2fs", + poll_elapsed, + ) + elif poll_elapsed > POLL_COMMAND_TIMEOUT: + logger.warning( + "SimpleX poll: /tail 50 slow response %.2fs", + poll_elapsed, + ) + logger.debug( + "SimpleX poll: got %d items in %.2fs (seen=%d)", + len(wrappers), poll_elapsed, len(self._seen_item_ids) + ) + for wrapper in wrappers: + chat_item = wrapper.get("chatItem") or {} + meta = chat_item.get("meta") or {} + item_key = self._item_key(wrapper) + if item_key is None: + continue + if item_key in self._seen_item_ids: + continue + item_ts = self._item_timestamp_epoch(wrapper) + if item_ts is not None and self._connected_at and item_ts < self._connected_at - 5: + self._seen_item_ids.add(item_key) + self._save_seen_items() + logger.info( + "SimpleX poll: marking stale pre-connect item seen: %s", + item_key, + ) + continue + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() + + status = (meta.get("itemStatus") or {}).get("type", "") + if status != "rcvNew": + continue + content = chat_item.get("content") or {} + if content.get("type") != "rcvMsgContent": + continue + logger.info( + "SimpleX poll: dispatching unread item %s", + item_key, + ) + self._save_seen_items() + asyncio.create_task(self._handle_new_chat_item(wrapper)) + except asyncio.CancelledError: + break + except Exception: + logger.exception("SimpleX poll: failed to poll unread items") + + async def _resolve_chat_target(self, chat_id: str) -> str: + """Resolve Hermes IDs to simplex-chat command targets. + + The command parser accepts display names (for example ``@'Elkim'`` or + ``#'group'``), not internal numeric contact/group IDs. Hermes keeps the + stable IDs, so resolve them via ``/chats`` before sending. + """ + resp = await self._command_once("/chats all") + chats = (resp or {}).get("resp", {}).get("chats", []) or [] + if chat_id.startswith("group:"): + wanted = str(chat_id[6:]) + for chat in chats: + group = ((chat.get("chatInfo") or {}).get("groupInfo") or {}) + if str(group.get("groupId")) == wanted: + name = group.get("localDisplayName") or (group.get("groupProfile") or {}).get("displayName") + if name: + return f"#'{_simplex_quote_name(name)}'" + if not wanted.isdigit(): + return f"#[{wanted}]" + else: + wanted = str(chat_id) + for chat in chats: + contact = ((chat.get("chatInfo") or {}).get("contact") or {}) + if str(contact.get("contactId")) == wanted: + name = contact.get("localDisplayName") or (contact.get("profile") or {}).get("displayName") + if name: + return f"@'{_simplex_quote_name(name)}'" + if not wanted.isdigit(): + return f"@[{wanted}]" + raise ValueError(f"SimpleX chat id not found: {chat_id}") + # ------------------------------------------------------------------ # Inbound event handling # ------------------------------------------------------------------ @@ -302,10 +534,11 @@ async def _handle_event(self, event: dict) -> None: return if resp_type == "newChatItem": - await self._handle_new_chat_item(event) - elif resp_type == "newChatItems": + await self._handle_new_chat_item(event.get("resp") or event) + elif resp_type in ("newChatItems", "chatItems"): # Batch variant — process each item - items = event.get("chatItems") or [] + payload = event.get("resp") or event + items = payload.get("chatItems") or [] for item_wrapper in items: await self._handle_new_chat_item(item_wrapper) # Ignore all other event types (delivery receipts, contact updates, etc.) @@ -502,20 +735,14 @@ async def send( metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send a text message to a contact or group.""" - corr_id = self._make_corr_id() - - if chat_id.startswith("group:"): - group_id = chat_id[6:] - cmd_str = f"#[{group_id}] {content}" - else: - cmd_str = f"@[{chat_id}] {content}" - - payload = { - "corrId": corr_id, - "cmd": cmd_str, - } - - await self._send_ws(payload) + target = await self._resolve_chat_target(chat_id) + if target.startswith("@[") or target.startswith("#["): + await self._send_ws({"corrId": self._make_corr_id(), "cmd": f"{target} {content}"}) + return SendResult(success=True) + resp = await self._command_once(f"{target} {content}") + resp_payload = (resp or {}).get("resp") or {} + if resp_payload.get("type") == "chatCmdError": + return SendResult(success=False, error=str(resp_payload.get("chatError"))) return SendResult(success=True) async def send_typing(self, chat_id: str, metadata=None) -> None: @@ -634,26 +861,67 @@ async def _standalone_send( return {"error": "websockets not installed. Run: pip install websockets"} extra = getattr(pconfig, "extra", {}) or {} - ws_url = os.getenv("SIMPLEX_WS_URL") or extra.get("ws_url", "ws://127.0.0.1:5225") + ws_url = os.getenv("SIMPLEX_WS_URL") or extra.get("ws_url", "") if not ws_url: return {"error": "SimpleX standalone send: SIMPLEX_WS_URL is required"} try: - if chat_id.startswith("group:"): - group_id = chat_id[6:] - cmd_str = f"#[{group_id}] {message}" - else: - cmd_str = f"@[{chat_id}] {message}" - - payload = { - "corrId": f"hermes-snd-{int(time.time() * 1000)}", - "cmd": cmd_str, - } - async with _wsclient.connect(ws_url, open_timeout=10, close_timeout=5) as ws: + # Resolve Hermes's stable numeric IDs to simplex-chat command names. + resolve_corr = f"hermes-resolve-{int(time.time() * 1000)}" + await ws.send(json.dumps({"corrId": resolve_corr, "cmd": "/chats all"})) + target = None + deadline = time.time() + 10 + while time.time() < deadline: + raw = await asyncio.wait_for(ws.recv(), timeout=max(0.1, deadline - time.time())) + msg = json.loads(raw) + if msg.get("corrId") != resolve_corr: + continue + chats = (msg.get("resp") or {}).get("chats", []) or [] + if chat_id.startswith("group:"): + wanted = str(chat_id[6:]) + for chat in chats: + group = ((chat.get("chatInfo") or {}).get("groupInfo") or {}) + if str(group.get("groupId")) == wanted: + name = group.get("localDisplayName") or (group.get("groupProfile") or {}).get("displayName") + if name: + target = f"#'{_simplex_quote_name(name)}'" + break + else: + wanted = str(chat_id) + for chat in chats: + contact = ((chat.get("chatInfo") or {}).get("contact") or {}) + if str(contact.get("contactId")) == wanted: + name = contact.get("localDisplayName") or (contact.get("profile") or {}).get("displayName") + if name: + target = f"@'{_simplex_quote_name(name)}'" + break + break + if not target: + if chat_id.startswith("group:"): + wanted = str(chat_id[6:]) + if not wanted.isdigit(): + target = f"#[{wanted}]" + elif not str(chat_id).isdigit(): + target = f"@[{chat_id}]" + if not target: + return {"error": f"SimpleX chat id not found: {chat_id}"} + + payload = { + "corrId": f"hermes-snd-{int(time.time() * 1000)}", + "cmd": f"{target} {message}", + } await ws.send(json.dumps(payload)) - # Give the daemon a moment to process the command before closing. - await asyncio.sleep(0.5) + deadline = time.time() + 10 + while time.time() < deadline: + raw = await asyncio.wait_for(ws.recv(), timeout=max(0.1, deadline - time.time())) + msg = json.loads(raw) + if msg.get("corrId") != payload["corrId"]: + continue + resp = msg.get("resp") or {} + if resp.get("type") == "chatCmdError": + return {"error": f"SimpleX send failed: {resp.get('chatError')}"} + break return {"success": True, "platform": "simplex", "chat_id": chat_id} except Exception as e: diff --git a/tests/gateway/test_simplex_plugin.py b/tests/gateway/test_simplex_plugin.py index 0b1b1b21a85ef..cc77720c39012 100644 --- a/tests/gateway/test_simplex_plugin.py +++ b/tests/gateway/test_simplex_plugin.py @@ -28,6 +28,9 @@ _is_image_ext = _simplex._is_image_ext _is_audio_ext = _simplex._is_audio_ext _CORR_PREFIX = _simplex._CORR_PREFIX +POLL_COMMAND_TIMEOUT = _simplex.POLL_COMMAND_TIMEOUT +POLL_CONNECT_TIMEOUT = _simplex.POLL_CONNECT_TIMEOUT +_simplex_quote_name = _simplex._simplex_quote_name # --------------------------------------------------------------------------- @@ -246,7 +249,131 @@ async def test_send_when_ws_not_connected_does_not_crash(): # --------------------------------------------------------------------------- -# 8. Inbound: filter own-echo by corrId prefix +# 8. Inbound: seen item bookkeeping +# --------------------------------------------------------------------------- + +def test_item_key_is_per_chat_not_global_item_id(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + direct = { + "chatInfo": {"type": "direct", "contact": {"contactId": 4}}, + "chatItem": {"meta": {"itemId": 7}}, + } + group = { + "chatInfo": {"type": "group", "groupInfo": {"groupId": 1}}, + "chatItem": {"meta": {"itemId": 7}}, + } + + assert adapter._item_key(direct) == "direct:4:7" + assert adapter._item_key(group) == "group:1:7" + + +def test_seen_items_persist_across_adapter_restart(tmp_path): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + + first = SimplexAdapter(cfg) + first._seen_items_path = tmp_path / "simplex_seen_items.json" + first._seen_item_ids.update({"direct:4:41", "group:1:39"}) + first._save_seen_items() + + second = SimplexAdapter(cfg) + second._seen_items_path = first._seen_items_path + second._load_seen_items() + + assert "direct:4:41" in second._seen_item_ids + assert "group:1:39" in second._seen_item_ids + + +def test_item_key_requires_chat_id(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + assert adapter._item_key({"chatInfo": {"type": "direct"}, "chatItem": {"meta": {"itemId": 7}}}) is None + + +def test_simplex_quote_name_escapes_single_quotes(): + assert _simplex_quote_name("Bob's Room") == "Bob\\'s Room" + + +@pytest.mark.asyncio +async def test_resolve_chat_target_uses_display_name(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + async def fake_command_once(cmd, **kwargs): + assert cmd == "/chats all" + return { + "resp": { + "chats": [ + {"chatInfo": {"contact": {"contactId": 4, "localDisplayName": "Elkim"}}}, + {"chatInfo": {"groupInfo": {"groupId": 1, "groupProfile": {"displayName": "Žofka_1"}}}}, + ] + } + } + + adapter._command_once = fake_command_once # type: ignore[method-assign] + + assert await adapter._resolve_chat_target("4") == "@'Elkim'" + assert await adapter._resolve_chat_target("group:1") == "#'Žofka_1'" + + +@pytest.mark.asyncio +async def test_resolve_chat_target_keeps_legacy_non_numeric_fallback(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + async def fake_command_once(cmd, **kwargs): + return {"resp": {"chats": []}} + + adapter._command_once = fake_command_once # type: ignore[method-assign] + + assert await adapter._resolve_chat_target("contact-42") == "@[contact-42]" + assert await adapter._resolve_chat_target("group:grp-99") == "#[grp-99]" + + +@pytest.mark.asyncio +async def test_poll_unread_uses_short_command_timeouts(monkeypatch): + """Polling is the latency fallback; it must not wait on 10s WS stalls.""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._running = True + + calls = [] + + async def fake_sleep(_seconds): + adapter._running = False + + async def fake_command_once(cmd, **kwargs): + calls.append((cmd, kwargs)) + return {"resp": {"chatItems": []}} + + monkeypatch.setattr(_simplex.asyncio, "sleep", fake_sleep) + adapter._command_once = fake_command_once # type: ignore[method-assign] + + await adapter._poll_unread_items() + + assert calls == [ + ( + "/tail 50", + { + "timeout": POLL_COMMAND_TIMEOUT, + "open_timeout": POLL_CONNECT_TIMEOUT, + }, + ) + ] + assert POLL_COMMAND_TIMEOUT <= 2.0 + assert POLL_CONNECT_TIMEOUT <= 2.0 + + +# --------------------------------------------------------------------------- +# 9. Inbound: filter own-echo by corrId prefix # --------------------------------------------------------------------------- @pytest.mark.asyncio From 8f8639d1d70e900e46da8b686fe07d7a87d5534e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=BDofka?= Date: Sun, 17 May 2026 11:34:24 +0200 Subject: [PATCH 2/4] fix(simplex): bound stale active sessions --- gateway/platforms/base.py | 50 ++++++++++++++ plugins/platforms/simplex/adapter.py | 61 +++++++++++++++-- tests/gateway/test_simplex_plugin.py | 97 +++++++++++++++++++++++++++- 3 files changed, 202 insertions(+), 6 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 96b56d29cc7d3..066a6b0fb1f17 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -15,6 +15,7 @@ import socket as _socket import subprocess import sys +import time import uuid from abc import ABC, abstractmethod from urllib.parse import urlsplit @@ -1296,6 +1297,7 @@ def __init__(self, config: PlatformConfig, platform: Platform): self._active_sessions: Dict[str, asyncio.Event] = {} self._pending_messages: Dict[str, MessageEvent] = {} self._session_tasks: Dict[str, asyncio.Task] = {} + self._session_started_at: Dict[str, float] = {} # Background message-processing tasks spawned by handle_message(). # Gateway shutdown cancels these so an old gateway instance doesn't keep # working on a task after --replace or manual restarts. @@ -2599,6 +2601,7 @@ def _release_session_guard( if guard is not None and current_guard is not guard: return del self._active_sessions[session_key] + self._session_started_at.pop(session_key, None) def _session_task_is_stale(self, session_key: str) -> bool: """Return True if the owner task for ``session_key`` is done/cancelled. @@ -2641,8 +2644,32 @@ def _heal_stale_session_lock(self, session_key: str) -> bool: self._active_sessions.pop(session_key, None) self._pending_messages.pop(session_key, None) self._session_tasks.pop(session_key, None) + self._session_started_at.pop(session_key, None) return True + def _active_session_age(self, session_key: str) -> float: + started = getattr(self, "_session_started_at", {}).get(session_key) + if not started: + return 0.0 + return max(0.0, time.time() - started) + + def _active_session_hard_timeout_seconds(self) -> float: + """Optional per-adapter hard cap for a busy session. + + Most platforms keep the default disabled and rely on cooperative + interrupt semantics. Polling transports such as SimpleX can opt in so a + fresh inbound item does not sit behind an old turn for tens of minutes + after that old turn stops reaching interrupt checkpoints. + """ + raw = getattr(self, "_max_active_session_seconds", None) + if raw is None: + raw = (getattr(self.config, "extra", {}) or {}).get("max_active_session_seconds", 0) + try: + value = float(raw or 0) + except (TypeError, ValueError): + return 0.0 + return max(0.0, value) + def _start_session_processing( self, event: MessageEvent, @@ -2659,6 +2686,7 @@ def _start_session_processing( """ guard = interrupt_event or asyncio.Event() self._active_sessions[session_key] = guard + self._session_started_at[session_key] = time.time() task = asyncio.create_task(self._process_message_background(event, session_key)) self._session_tasks[session_key] = task @@ -2668,6 +2696,7 @@ def _start_session_processing( # Tests stub create_task() with lightweight sentinels that are not # hashable and do not support lifecycle callbacks. self._session_tasks.pop(session_key, None) + self._session_started_at.pop(session_key, None) self._release_session_guard(session_key, guard=guard) return False if hasattr(task, "add_done_callback"): @@ -2846,6 +2875,24 @@ async def handle_message(self, event: MessageEvent) -> None: if session_key in self._active_sessions: self._heal_stale_session_lock(session_key) + # Optional hard-timeout safety valve for polling transports. A + # cooperative interrupt only helps once the old agent reaches an + # interrupt checkpoint; if an opted-in adapter has exceeded its cap, + # cancel the stale turn before queueing this fresh user message behind + # it. + if session_key in self._active_sessions: + hard_timeout = self._active_session_hard_timeout_seconds() + active_age = self._active_session_age(session_key) + if hard_timeout and active_age > hard_timeout: + logger.warning( + "[%s] Active session %s exceeded %.0fs (age %.0fs); cancelling stale turn before processing new message", + self.name, + session_key, + hard_timeout, + active_age, + ) + await self.cancel_session_processing(session_key) + # Check if there's already an active handler for this session if session_key in self._active_sessions: # Certain commands must bypass the active-session guard and be @@ -3362,6 +3409,7 @@ async def _stop_typing_task() -> None: # Hand ownership of the session to the drain task so # stale-lock detection keeps working while it runs. self._session_tasks[session_key] = drain_task + self._session_started_at[session_key] = time.time() try: self._background_tasks.add(drain_task) drain_task.add_done_callback(self._background_tasks.discard) @@ -3474,6 +3522,7 @@ async def _stop_typing_task() -> None: # Hand ownership of the session to the drain task so stale-lock # detection keeps working while it runs. self._session_tasks[session_key] = drain_task + self._session_started_at[session_key] = time.time() try: self._background_tasks.add(drain_task) drain_task.add_done_callback(self._background_tasks.discard) @@ -3503,6 +3552,7 @@ async def _stop_typing_task() -> None: current_task = asyncio.current_task() if current_task is not None and self._session_tasks.get(session_key) is current_task: del self._session_tasks[session_key] + self._session_started_at.pop(session_key, None) self._release_session_guard(session_key, guard=interrupt_event) async def cancel_background_tasks(self) -> None: diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index f7992b9e07b3a..61ef501639a4a 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -73,6 +73,7 @@ # the 50–60s "SimpleX feels dead" delay after a few consecutive stalls. POLL_COMMAND_TIMEOUT = 2.0 POLL_CONNECT_TIMEOUT = 2.0 +SIMPLEX_ACTIVE_SESSION_MAX_SECONDS = 300.0 # Correlation ID prefix for requests we send so we can ignore our own echoes. _CORR_PREFIX = "hermes-" @@ -150,6 +151,17 @@ def __init__(self, config: PlatformConfig, **kwargs): self._seen_item_ids: set = set() self._seen_items_path = get_hermes_home() / "simplex_seen_items.json" self._connected_at = 0.0 + self._poll_dispatch_tasks: set[asyncio.Task] = set() + # SimpleX push/poll delivery should feel like chat, not a mailbox. If + # an old SimpleX turn stops reaching cooperative interrupt checkpoints, + # BasePlatformAdapter can cancel it when a fresh SimpleX message arrives + # instead of queueing that message for tens of minutes. + try: + self._max_active_session_seconds = float( + extra.get("max_active_session_seconds", SIMPLEX_ACTIVE_SESSION_MAX_SECONDS) + ) + except (TypeError, ValueError): + self._max_active_session_seconds = SIMPLEX_ACTIVE_SESSION_MAX_SECONDS # Track sent correlation IDs to filter echoes self._pending_corr_ids: set = set() @@ -222,6 +234,12 @@ async def disconnect(self) -> None: except asyncio.CancelledError: pass + for task in list(self._poll_dispatch_tasks): + task.cancel() + if self._poll_dispatch_tasks: + await asyncio.gather(*self._poll_dispatch_tasks, return_exceptions=True) + self._poll_dispatch_tasks.clear() + for task in self._typing_tasks.values(): task.cancel() self._typing_tasks.clear() @@ -467,27 +485,60 @@ async def _poll_unread_items(self) -> None: item_key, ) continue - self._seen_item_ids.add(item_key) - if len(self._seen_item_ids) > 1000: - self._save_seen_items() - status = (meta.get("itemStatus") or {}).get("type", "") if status != "rcvNew": + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() continue content = chat_item.get("content") or {} if content.get("type") != "rcvMsgContent": + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() continue logger.info( "SimpleX poll: dispatching unread item %s", item_key, ) + self._seen_item_ids.add(item_key) self._save_seen_items() - asyncio.create_task(self._handle_new_chat_item(wrapper)) + self._dispatch_polled_item(wrapper, item_key) except asyncio.CancelledError: break except Exception: logger.exception("SimpleX poll: failed to poll unread items") + def _dispatch_polled_item(self, wrapper: dict, item_key: str) -> asyncio.Task: + """Dispatch a polled item and keep enough state to observe failures.""" + started = time.time() + + async def _runner() -> None: + try: + await self._handle_new_chat_item(wrapper) + elapsed = time.time() - started + if elapsed > 5.0: + logger.warning( + "SimpleX poll: item %s dispatch took %.2fs", + item_key, + elapsed, + ) + else: + logger.info( + "SimpleX poll: item %s handed to gateway in %.2fs", + item_key, + elapsed, + ) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("SimpleX poll: item %s dispatch failed", item_key) + + task = asyncio.create_task(_runner(), name=f"simplex-poll-dispatch:{item_key}") + self._poll_dispatch_tasks.add(task) + task.add_done_callback(self._poll_dispatch_tasks.discard) + return task + async def _resolve_chat_target(self, chat_id: str) -> str: """Resolve Hermes IDs to simplex-chat command targets. diff --git a/tests/gateway/test_simplex_plugin.py b/tests/gateway/test_simplex_plugin.py index cc77720c39012..d4936b763cadf 100644 --- a/tests/gateway/test_simplex_plugin.py +++ b/tests/gateway/test_simplex_plugin.py @@ -30,6 +30,7 @@ _CORR_PREFIX = _simplex._CORR_PREFIX POLL_COMMAND_TIMEOUT = _simplex.POLL_COMMAND_TIMEOUT POLL_CONNECT_TIMEOUT = _simplex.POLL_CONNECT_TIMEOUT +SIMPLEX_ACTIVE_SESSION_MAX_SECONDS = _simplex.SIMPLEX_ACTIVE_SESSION_MAX_SECONDS _simplex_quote_name = _simplex._simplex_quote_name @@ -67,8 +68,9 @@ def test_check_requirements_true_when_configured(monkeypatch): assert check_requirements() is websockets_present -def test_validate_config_uses_env_or_extra(): +def test_validate_config_uses_env_or_extra(monkeypatch): from gateway.config import PlatformConfig + monkeypatch.delenv("SIMPLEX_WS_URL", raising=False) # Empty extra + no env → invalid cfg = PlatformConfig(enabled=True) assert validate_config(cfg) is False @@ -372,6 +374,99 @@ async def fake_command_once(cmd, **kwargs): assert POLL_CONNECT_TIMEOUT <= 2.0 +def test_simplex_adapter_opts_into_active_session_hard_timeout(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + assert adapter._max_active_session_seconds == SIMPLEX_ACTIVE_SESSION_MAX_SECONDS + assert adapter._active_session_hard_timeout_seconds() == SIMPLEX_ACTIVE_SESSION_MAX_SECONDS + + +def test_simplex_adapter_respects_configured_active_session_hard_timeout(): + from gateway.config import PlatformConfig + cfg = PlatformConfig( + enabled=True, + extra={"ws_url": "ws://localhost:5225", "max_active_session_seconds": 12}, + ) + adapter = SimplexAdapter(cfg) + + assert adapter._active_session_hard_timeout_seconds() == 12.0 + + +@pytest.mark.asyncio +async def test_dispatch_polled_item_tracks_task_and_logs_dispatch(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + gate = _simplex.asyncio.Event() + handled = [] + + async def fake_handle(wrapper): + handled.append(wrapper) + await gate.wait() + + adapter._handle_new_chat_item = fake_handle # type: ignore[method-assign] + task = adapter._dispatch_polled_item({"x": 1}, "direct:4:99") + + await _simplex.asyncio.sleep(0) + assert task in adapter._poll_dispatch_tasks + assert handled == [{"x": 1}] + + gate.set() + await task + assert task not in adapter._poll_dispatch_tasks + + +@pytest.mark.asyncio +async def test_stale_active_simplex_session_is_cancelled_for_fresh_message(): + from gateway.config import PlatformConfig + from gateway.platforms.base import MessageType + + cfg = PlatformConfig( + enabled=True, + extra={"ws_url": "ws://localhost:5225", "max_active_session_seconds": 0.01}, + ) + adapter = SimplexAdapter(cfg) + adapter._max_active_session_seconds = 0.01 + + started = _simplex.asyncio.Event() + cancelled = _simplex.asyncio.Event() + handled_texts = [] + + async def handler(event): + handled_texts.append(event.text) + if event.text == "old": + started.set() + try: + await _simplex.asyncio.sleep(60) + except _simplex.asyncio.CancelledError: + cancelled.set() + raise + return "" + + adapter.set_message_handler(handler) + source = adapter.build_source( + chat_id="4", + chat_name="Elkim", + chat_type="dm", + user_id="4", + user_name="Elkim", + ) + old = _simplex.MessageEvent(source=source, text="old", message_type=MessageType.TEXT) + fresh = _simplex.MessageEvent(source=source, text="fresh", message_type=MessageType.TEXT) + + await adapter.handle_message(old) + await _simplex.asyncio.wait_for(started.wait(), timeout=1) + await _simplex.asyncio.sleep(0.02) + await adapter.handle_message(fresh) + await _simplex.asyncio.wait_for(cancelled.wait(), timeout=1) + await _simplex.asyncio.sleep(0) + + assert handled_texts[:2] == ["old", "fresh"] + + # --------------------------------------------------------------------------- # 9. Inbound: filter own-echo by corrId prefix # --------------------------------------------------------------------------- From 2d8522ccabd669d79ff7fddccdd389fc8cf58f15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=BDofka?= Date: Sun, 17 May 2026 11:53:59 +0200 Subject: [PATCH 3/4] fix(simplex): make polling recovery visible and bounded Improve the SimpleX fallback path when daemon push events are stale or missing. Keep polling on a short wall-clock budget, dispatch polled batches in daemon order, mark polled inbound items read, avoid seeding unread inbound text as stale history, and show a delayed processing notice for slow SimpleX turns. --- plugins/platforms/simplex/adapter.py | 330 ++++++++++++++++++++++----- tests/gateway/test_simplex_plugin.py | 295 ++++++++++++++++++++++++ 2 files changed, 572 insertions(+), 53 deletions(-) diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index 61ef501639a4a..a75a3e1d77f41 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -73,7 +73,10 @@ # the 50–60s "SimpleX feels dead" delay after a few consecutive stalls. POLL_COMMAND_TIMEOUT = 2.0 POLL_CONNECT_TIMEOUT = 2.0 +POLL_WALL_TIMEOUT = 3.5 +POLL_STALL_WARN_SECONDS = 5.0 SIMPLEX_ACTIVE_SESSION_MAX_SECONDS = 300.0 +SIMPLEX_PROCESSING_NOTICE_DELAY = 5.0 # Correlation ID prefix for requests we send so we can ignore our own echoes. _CORR_PREFIX = "hermes-" @@ -152,6 +155,15 @@ def __init__(self, config: PlatformConfig, **kwargs): self._seen_items_path = get_hermes_home() / "simplex_seen_items.json" self._connected_at = 0.0 self._poll_dispatch_tasks: set[asyncio.Task] = set() + self._read_receipt_tasks: set[asyncio.Task] = set() + self._last_poll_started_at = 0.0 + self._processing_notice_tasks: Dict[str, asyncio.Task] = {} + try: + self._processing_notice_delay = float( + extra.get("processing_notice_delay", SIMPLEX_PROCESSING_NOTICE_DELAY) + ) + except (TypeError, ValueError): + self._processing_notice_delay = SIMPLEX_PROCESSING_NOTICE_DELAY # SimpleX push/poll delivery should feel like chat, not a mailbox. If # an old SimpleX turn stops reaching cooperative interrupt checkpoints, # BasePlatformAdapter can cancel it when a fresh SimpleX message arrives @@ -240,6 +252,18 @@ async def disconnect(self) -> None: await asyncio.gather(*self._poll_dispatch_tasks, return_exceptions=True) self._poll_dispatch_tasks.clear() + for task in list(self._read_receipt_tasks): + task.cancel() + if self._read_receipt_tasks: + await asyncio.gather(*self._read_receipt_tasks, return_exceptions=True) + self._read_receipt_tasks.clear() + + for task in list(self._processing_notice_tasks.values()): + task.cancel() + if self._processing_notice_tasks: + await asyncio.gather(*self._processing_notice_tasks.values(), return_exceptions=True) + self._processing_notice_tasks.clear() + for task in self._typing_tasks.values(): task.cancel() self._typing_tasks.clear() @@ -274,6 +298,51 @@ def _item_key(self, wrapper: dict) -> Optional[str]: return None return f"{prefix}:{chat_id}:{item_id}" + def _chat_ref(self, wrapper: dict) -> Optional[str]: + """Return the simplex-chat API chat reference for a wrapper.""" + chat_info = wrapper.get("chatInfo") or wrapper.get("chat") or {} + chat_type = chat_info.get("type") or "" + if chat_type in ("group", "groupInfo"): + group_info = chat_info.get("groupInfo") or chat_info.get("group") or {} + chat_id = group_info.get("groupId") or group_info.get("id") or "" + return f"#{chat_id}" if chat_id else None + contact_info = chat_info.get("contact") or {} + chat_id = contact_info.get("contactId") or contact_info.get("id") or "" + return f"@{chat_id}" if chat_id else None + + def _mark_item_read_soon(self, wrapper: dict) -> None: + """Tell simplex-chat this inbound item was seen without blocking handling.""" + chat_ref = self._chat_ref(wrapper) + chat_item = wrapper.get("chatItem") or wrapper.get("item") or {} + item_id = (chat_item.get("meta") or {}).get("itemId") + if not chat_ref or item_id is None: + return + + async def _runner() -> None: + resp = await self._command_once( + f"/_read chat items {chat_ref} {item_id}", + timeout=1.0, + open_timeout=1.0, + wall_timeout=2.0, + ) + resp_type = (resp or {}).get("resp", {}).get("type") + if resp_type not in ("itemsReadForChat", "cmdOk"): + logger.debug("SimpleX: read receipt command returned %s", resp_type) + + task = asyncio.create_task(_runner(), name=f"simplex-mark-read:{chat_ref}:{item_id}") + self._read_receipt_tasks.add(task) + + def _cleanup(done: asyncio.Task) -> None: + self._read_receipt_tasks.discard(done) + try: + done.result() + except asyncio.CancelledError: + pass + except Exception: + logger.debug("SimpleX: failed to mark item read", exc_info=True) + + task.add_done_callback(_cleanup) + def _item_timestamp_epoch(self, wrapper: dict) -> Optional[float]: chat_item = wrapper.get("chatItem") or wrapper.get("item") or {} meta = chat_item.get("meta") or {} @@ -285,6 +354,25 @@ def _item_timestamp_epoch(self, wrapper: dict) -> Optional[float]: except (ValueError, AttributeError): return None + + def _is_unread_inbound_text(self, wrapper: dict) -> bool: + """Return True for inbound unread text that still needs dispatch. + + These items must never be treated as restart seed/stale history: if an + unread DM sat in the daemon while Hermes was down or the poller was + wedged, the recovery path is to process it, not consume it silently. + """ + chat_item = wrapper.get("chatItem") or wrapper.get("item") or {} + meta = chat_item.get("meta") or {} + status = (meta.get("itemStatus") or {}).get("type", "") + if status != "rcvNew": + return False + content = chat_item.get("content") or {} + if content.get("type") != "rcvMsgContent": + return False + msg_content = content.get("msgContent") or {} + return bool(msg_content) + def _load_seen_items(self) -> None: try: data = json.loads(Path(self._seen_items_path).read_text(encoding="utf-8")) @@ -390,20 +478,14 @@ async def _health_monitor(self) -> None: # Polling fallback # ------------------------------------------------------------------ - async def _command_once( + async def _command_once_impl( self, cmd: str, *, timeout: float = 10.0, open_timeout: float = 10.0, ) -> Optional[dict]: - """Run one simplex-chat command over an ephemeral WebSocket. - - The daemon does not reliably push ``newChatItem`` events to every - persistent WebSocket client in all versions/modes. A tiny polling - fallback makes inbound DM/group delivery deterministic while keeping - the persistent listener for installations where push events work. - """ + """Run one simplex-chat command over an ephemeral WebSocket.""" import websockets as _wsclient corr_id = self._make_corr_id() @@ -425,16 +507,77 @@ async def _command_once( self._pending_corr_ids.discard(corr_id) return None + async def _command_once( + self, + cmd: str, + *, + timeout: float = 10.0, + open_timeout: float = 10.0, + wall_timeout: Optional[float] = None, + ) -> Optional[dict]: + """Run one simplex-chat command with a hard wall-clock bound. + + The daemon does not reliably push ``newChatItem`` events to every + persistent WebSocket client in all versions/modes. Polling is the + reliability path, so even websocket close/connect edge cases must not + stall the poll loop indefinitely. + """ + effective_wall_timeout = wall_timeout + if effective_wall_timeout is None: + effective_wall_timeout = max(open_timeout + timeout + 1.5, timeout + 1.5) + task = asyncio.create_task( + self._command_once_impl(cmd, timeout=timeout, open_timeout=open_timeout), + name=f"simplex-command:{cmd[:32]}", + ) + try: + return await asyncio.wait_for(asyncio.shield(task), timeout=effective_wall_timeout) + except asyncio.TimeoutError: + logger.warning( + "SimpleX command %r exceeded wall timeout %.1fs", + cmd, + effective_wall_timeout, + ) + task.cancel() + + def _log_late_failure(done: asyncio.Task) -> None: + try: + done.result() + except asyncio.CancelledError: + pass + except Exception: + logger.debug( + "SimpleX command %r background cleanup failed after timeout", + cmd, + exc_info=True, + ) + + task.add_done_callback(_log_late_failure) + return None + async def _seed_seen_items(self) -> None: """Remember current history so a gateway restart doesn't answer old mail.""" try: resp = await self._command_once("/tail 50") changed = False + skipped_unread = 0 for wrapper in (resp or {}).get("resp", {}).get("chatItems", []) or []: item_key = self._item_key(wrapper) - if item_key is not None: - self._seen_item_ids.add(item_key) - changed = True + if item_key is None: + continue + if self._is_unread_inbound_text(wrapper): + # Do not eat fresh/missed user mail on startup. Persistent + # seen state already prevents old processed messages from + # replaying; unread inbound text should flow through the + # normal poll dispatch path on the first tick after connect. + skipped_unread += 1 + continue + self._seen_item_ids.add(item_key) + changed = True + if skipped_unread: + logger.info( + "SimpleX poll: seed left %d unread inbound text item(s) for dispatch", + skipped_unread, + ) if changed: self._save_seen_items() logger.info("SimpleX poll: seeded %d seen chat items", len(self._seen_item_ids)) @@ -447,10 +590,19 @@ async def _poll_unread_items(self) -> None: await asyncio.sleep(POLL_INTERVAL) try: start = time.time() + if self._last_poll_started_at: + poll_gap = start - self._last_poll_started_at + if poll_gap > POLL_STALL_WARN_SECONDS: + logger.warning( + "SimpleX poll: loop stalled for %.2fs before next /tail", + poll_gap, + ) + self._last_poll_started_at = start resp = await self._command_once( "/tail 50", timeout=POLL_COMMAND_TIMEOUT, open_timeout=POLL_CONNECT_TIMEOUT, + wall_timeout=POLL_WALL_TIMEOUT, ) poll_elapsed = time.time() - start wrappers = (resp or {}).get("resp", {}).get("chatItems", []) or [] @@ -468,6 +620,7 @@ async def _poll_unread_items(self) -> None: "SimpleX poll: got %d items in %.2fs (seen=%d)", len(wrappers), poll_elapsed, len(self._seen_item_ids) ) + dispatch_items: list[tuple[dict, str]] = [] for wrapper in wrappers: chat_item = wrapper.get("chatItem") or {} meta = chat_item.get("meta") or {} @@ -476,65 +629,88 @@ async def _poll_unread_items(self) -> None: continue if item_key in self._seen_item_ids: continue - item_ts = self._item_timestamp_epoch(wrapper) - if item_ts is not None and self._connected_at and item_ts < self._connected_at - 5: - self._seen_item_ids.add(item_key) - self._save_seen_items() - logger.info( - "SimpleX poll: marking stale pre-connect item seen: %s", - item_key, - ) - continue - status = (meta.get("itemStatus") or {}).get("type", "") - if status != "rcvNew": - self._seen_item_ids.add(item_key) - if len(self._seen_item_ids) > 1000: + if not self._is_unread_inbound_text(wrapper): + item_ts = self._item_timestamp_epoch(wrapper) + if item_ts is not None and self._connected_at and item_ts < self._connected_at - 5: + self._seen_item_ids.add(item_key) self._save_seen_items() - continue - content = chat_item.get("content") or {} - if content.get("type") != "rcvMsgContent": - self._seen_item_ids.add(item_key) - if len(self._seen_item_ids) > 1000: - self._save_seen_items() - continue + logger.info( + "SimpleX poll: marking stale pre-connect item seen: %s", + item_key, + ) + continue + status = (meta.get("itemStatus") or {}).get("type", "") + if status != "rcvNew": + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() + continue + content = chat_item.get("content") or {} + if content.get("type") != "rcvMsgContent": + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() + continue logger.info( "SimpleX poll: dispatching unread item %s", item_key, ) self._seen_item_ids.add(item_key) self._save_seen_items() - self._dispatch_polled_item(wrapper, item_key) + dispatch_items.append((wrapper, item_key)) + if dispatch_items: + self._dispatch_polled_items(dispatch_items) except asyncio.CancelledError: break except Exception: logger.exception("SimpleX poll: failed to poll unread items") def _dispatch_polled_item(self, wrapper: dict, item_key: str) -> asyncio.Task: - """Dispatch a polled item and keep enough state to observe failures.""" + """Dispatch one polled item asynchronously. + + Kept as a small compatibility wrapper for tests and callers that hand us + a single item. The poll loop itself uses ``_dispatch_polled_items`` so a + burst from one `/tail` response is handed to the gateway in daemon order + rather than racing independent tasks against each other. + """ + return self._dispatch_polled_items([(wrapper, item_key)]) + + def _dispatch_polled_items(self, items: list[tuple[dict, str]]) -> asyncio.Task: + """Dispatch polled items in order while keeping the poll loop free.""" + first_key = items[0][1] if items else "empty" started = time.time() async def _runner() -> None: - try: - await self._handle_new_chat_item(wrapper) - elapsed = time.time() - started - if elapsed > 5.0: - logger.warning( - "SimpleX poll: item %s dispatch took %.2fs", - item_key, - elapsed, - ) - else: - logger.info( - "SimpleX poll: item %s handed to gateway in %.2fs", - item_key, - elapsed, - ) - except asyncio.CancelledError: - raise - except Exception: - logger.exception("SimpleX poll: item %s dispatch failed", item_key) + for wrapper, item_key in items: + item_started = time.time() + try: + await self._handle_new_chat_item(wrapper) + elapsed = time.time() - item_started + if elapsed > 5.0: + logger.warning( + "SimpleX poll: item %s dispatch took %.2fs", + item_key, + elapsed, + ) + else: + logger.info( + "SimpleX poll: item %s handed to gateway in %.2fs", + item_key, + elapsed, + ) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("SimpleX poll: item %s dispatch failed", item_key) + batch_elapsed = time.time() - started + if len(items) > 1: + logger.info( + "SimpleX poll: ordered batch of %d items handed to gateway in %.2fs", + len(items), + batch_elapsed, + ) - task = asyncio.create_task(_runner(), name=f"simplex-poll-dispatch:{item_key}") + task = asyncio.create_task(_runner(), name=f"simplex-poll-dispatch:{first_key}") self._poll_dispatch_tasks.add(task) task.add_done_callback(self._poll_dispatch_tasks.discard) return task @@ -613,6 +789,11 @@ async def _handle_new_chat_item(self, wrapper: dict) -> None: if direction in ("sndSent", "sndSentDirect", "sndSentViaProxy", "sndNew"): return + # SimpleX does not send read receipts for bot-polled items unless the + # local chat item is explicitly marked read. Do this before the LLM turn + # so the sender immediately sees that the daemon/gateway consumed it. + self._mark_item_read_soon(wrapper) + # Determine chat type and IDs chat_type_raw = chat_info.get("type", "") is_group = chat_type_raw in ("group", "groupInfo") @@ -750,6 +931,49 @@ async def _fetch_file(self, file_id: Any, file_name: str) -> Optional[str]: return cache_document_from_bytes(data, file_name) return None + # ------------------------------------------------------------------ + # Processing visibility + # ------------------------------------------------------------------ + + def _processing_notice_key(self, event: MessageEvent) -> str: + item_key = self._item_key(event.raw_message or {}) if isinstance(event.raw_message, dict) else None + return item_key or f"{event.source.chat_id}:{event.message_id or id(event)}" + + async def on_processing_start(self, event: MessageEvent) -> None: + """Send a delayed visible notice because SimpleX has no typing indicator.""" + if self._processing_notice_delay <= 0: + return + key = self._processing_notice_key(event) + + async def _notice() -> None: + try: + await asyncio.sleep(self._processing_notice_delay) + await self.send( + event.source.chat_id, + "Still here — SimpleX has no typing indicator, but I’m working on it.", + ) + logger.info("SimpleX visibility: sent delayed processing notice for %s", key) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.debug("SimpleX visibility notice failed for %s: %s", key, exc) + + old = self._processing_notice_tasks.pop(key, None) + if old is not None: + old.cancel() + task = asyncio.create_task(_notice(), name=f"simplex-processing-notice:{key}") + self._processing_notice_tasks[key] = task + + async def on_processing_complete(self, event: MessageEvent, outcome) -> None: + key = self._processing_notice_key(event) + task = self._processing_notice_tasks.pop(key, None) + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + # ------------------------------------------------------------------ # Outbound messages # ------------------------------------------------------------------ diff --git a/tests/gateway/test_simplex_plugin.py b/tests/gateway/test_simplex_plugin.py index d4936b763cadf..c21e6d18d018a 100644 --- a/tests/gateway/test_simplex_plugin.py +++ b/tests/gateway/test_simplex_plugin.py @@ -30,7 +30,10 @@ _CORR_PREFIX = _simplex._CORR_PREFIX POLL_COMMAND_TIMEOUT = _simplex.POLL_COMMAND_TIMEOUT POLL_CONNECT_TIMEOUT = _simplex.POLL_CONNECT_TIMEOUT +POLL_WALL_TIMEOUT = _simplex.POLL_WALL_TIMEOUT +POLL_STALL_WARN_SECONDS = _simplex.POLL_STALL_WARN_SECONDS SIMPLEX_ACTIVE_SESSION_MAX_SECONDS = _simplex.SIMPLEX_ACTIVE_SESSION_MAX_SECONDS +SIMPLEX_PROCESSING_NOTICE_DELAY = _simplex.SIMPLEX_PROCESSING_NOTICE_DELAY _simplex_quote_name = _simplex._simplex_quote_name @@ -270,6 +273,45 @@ def test_item_key_is_per_chat_not_global_item_id(): assert adapter._item_key(direct) == "direct:4:7" assert adapter._item_key(group) == "group:1:7" + assert adapter._chat_ref(direct) == "@4" + assert adapter._chat_ref(group) == "#1" + + +@pytest.mark.asyncio +async def test_handle_new_chat_item_marks_inbound_item_read_without_blocking(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + calls = [] + handled = [] + + async def fake_command_once(cmd, **kwargs): + calls.append((cmd, kwargs)) + return {"resp": {"type": "itemsReadForChat"}} + + async def handler(event): + handled.append(event.text) + return "" + + adapter._command_once = fake_command_once # type: ignore[method-assign] + adapter.set_message_handler(handler) + wrapper = { + "chatInfo": {"type": "direct", "contact": {"contactId": 4, "localDisplayName": "Elkim"}}, + "chatItem": { + "meta": {"itemId": 95, "itemStatus": {"type": "rcvNew"}, "itemTs": "2026-05-17T10:24:35Z"}, + "content": {"type": "rcvMsgContent", "msgContent": {"type": "text", "text": "ping"}}, + }, + } + + await adapter._handle_new_chat_item(wrapper) + await _simplex.asyncio.sleep(0) + + assert handled == ["ping"] + assert calls == [( + "/_read chat items @4 95", + {"timeout": 1.0, "open_timeout": 1.0, "wall_timeout": 2.0}, + )] def test_seen_items_persist_across_adapter_restart(tmp_path): @@ -339,6 +381,79 @@ async def fake_command_once(cmd, **kwargs): assert await adapter._resolve_chat_target("group:grp-99") == "#[grp-99]" +@pytest.mark.asyncio +async def test_seed_seen_items_does_not_consume_unread_inbound_text(tmp_path): + """Startup recovery must not silently eat missed SimpleX DMs.""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._seen_items_path = tmp_path / "simplex_seen_items.json" + + unread = { + "chatInfo": {"type": "direct", "contact": {"contactId": 4}}, + "chatItem": { + "meta": {"itemId": 100, "itemStatus": {"type": "rcvNew"}, "itemTs": "2026-05-17T10:43:07Z"}, + "content": {"type": "rcvMsgContent", "msgContent": {"type": "text", "text": "latency test 3"}}, + }, + } + sent = { + "chatInfo": {"type": "direct", "contact": {"contactId": 4}}, + "chatItem": { + "meta": {"itemId": 99, "itemStatus": {"type": "sndRcvd"}, "itemTs": "2026-05-17T10:51:47Z"}, + "content": {"type": "sndMsgContent", "msgContent": {"type": "text", "text": "shutdown"}}, + }, + } + + async def fake_command_once(_cmd, **_kwargs): + return {"resp": {"chatItems": [unread, sent]}} + + adapter._command_once = fake_command_once # type: ignore[method-assign] + await adapter._seed_seen_items() + + assert "direct:4:100" not in adapter._seen_item_ids + assert "direct:4:99" in adapter._seen_item_ids + + +@pytest.mark.asyncio +async def test_poll_dispatches_preconnect_unread_inbound_text(monkeypatch, tmp_path): + """Unread user text remains actionable even if its timestamp predates reconnect.""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._seen_items_path = tmp_path / "simplex_seen_items.json" + adapter._running = True + adapter._connected_at = 1_779_012_000.0 # after the item timestamp + handled = [] + + unread = { + "chatInfo": {"type": "direct", "contact": {"contactId": 4}}, + "chatItem": { + "meta": {"itemId": 100, "itemStatus": {"type": "rcvNew"}, "itemTs": "2026-05-17T10:43:07Z"}, + "content": {"type": "rcvMsgContent", "msgContent": {"type": "text", "text": "latency test 3"}}, + }, + } + + async def fake_sleep(_seconds): + adapter._running = False + + async def fake_command_once(_cmd, **_kwargs): + return {"resp": {"chatItems": [unread]}} + + async def fake_handle(wrapper): + handled.append(wrapper["chatItem"]["meta"]["itemId"]) + + monkeypatch.setattr(_simplex.asyncio, "sleep", fake_sleep) + adapter._command_once = fake_command_once # type: ignore[method-assign] + adapter._handle_new_chat_item = fake_handle # type: ignore[method-assign] + + await adapter._poll_unread_items() + for task in list(adapter._poll_dispatch_tasks): + await task + + assert handled == [100] + assert "direct:4:100" in adapter._seen_item_ids + + @pytest.mark.asyncio async def test_poll_unread_uses_short_command_timeouts(monkeypatch): """Polling is the latency fallback; it must not wait on 10s WS stalls.""" @@ -367,11 +482,165 @@ async def fake_command_once(cmd, **kwargs): { "timeout": POLL_COMMAND_TIMEOUT, "open_timeout": POLL_CONNECT_TIMEOUT, + "wall_timeout": POLL_WALL_TIMEOUT, }, ) ] assert POLL_COMMAND_TIMEOUT <= 2.0 assert POLL_CONNECT_TIMEOUT <= 2.0 + assert POLL_WALL_TIMEOUT <= 3.5 + + +@pytest.mark.asyncio +async def test_command_once_has_hard_wall_timeout(monkeypatch, caplog): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + async def stuck_impl(*_args, **_kwargs): + await _simplex.asyncio.sleep(60) + + adapter._command_once_impl = stuck_impl # type: ignore[method-assign] + + with caplog.at_level("WARNING"): + result = await adapter._command_once( + "/tail 50", + timeout=10, + open_timeout=10, + wall_timeout=0.01, + ) + + assert result is None + assert "exceeded wall timeout" in caplog.text + + +@pytest.mark.asyncio +async def test_command_once_wall_timeout_does_not_wait_for_slow_cancellation(caplog): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + cleanup_started = _simplex.asyncio.Event() + + async def stubborn_impl(*_args, **_kwargs): + try: + await _simplex.asyncio.sleep(60) + except _simplex.asyncio.CancelledError: + cleanup_started.set() + await _simplex.asyncio.sleep(0.2) + raise + + adapter._command_once_impl = stubborn_impl # type: ignore[method-assign] + + started = _simplex.time.time() + with caplog.at_level("WARNING"): + result = await adapter._command_once( + "/tail 50", + timeout=10, + open_timeout=10, + wall_timeout=0.01, + ) + elapsed = _simplex.time.time() - started + + assert result is None + assert elapsed < 0.1 + assert "exceeded wall timeout" in caplog.text + await _simplex.asyncio.wait_for(cleanup_started.wait(), timeout=1) + await _simplex.asyncio.sleep(0.25) + + +@pytest.mark.asyncio +async def test_poll_loop_logs_stall_gap(monkeypatch, caplog): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + adapter._running = True + adapter._last_poll_started_at = 100.0 + + async def fake_sleep(_seconds): + adapter._running = False + + async def fake_command_once(_cmd, **_kwargs): + return {"resp": {"chatItems": []}} + + times = iter([100.0 + POLL_STALL_WARN_SECONDS + 1.0, 100.1]) + monkeypatch.setattr(_simplex.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(_simplex.time, "time", lambda: next(times, 100.1)) + adapter._command_once = fake_command_once # type: ignore[method-assign] + + with caplog.at_level("WARNING"): + await adapter._poll_unread_items() + + assert "loop stalled" in caplog.text + + +@pytest.mark.asyncio +async def test_processing_notice_is_delayed_and_cancelled_before_send(monkeypatch): + from gateway.config import PlatformConfig + from gateway.platforms.base import MessageType, ProcessingOutcome + + cfg = PlatformConfig( + enabled=True, + extra={"ws_url": "ws://localhost:5225", "processing_notice_delay": 10}, + ) + adapter = SimplexAdapter(cfg) + sent = [] + + async def fake_send(chat_id, content, **_kwargs): + sent.append((chat_id, content)) + from gateway.platforms.base import SendResult + return SendResult(success=True) + + adapter.send = fake_send # type: ignore[method-assign] + source = adapter.build_source( + chat_id="4", + chat_name="Elkim", + chat_type="dm", + user_id="4", + user_name="Elkim", + ) + event = _simplex.MessageEvent(source=source, text="slow", message_type=MessageType.TEXT) + + await adapter.on_processing_start(event) + assert adapter._processing_notice_tasks + await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS) + await _simplex.asyncio.sleep(0) + + assert sent == [] + assert adapter._processing_notice_tasks == {} + + +@pytest.mark.asyncio +async def test_processing_notice_sends_for_slow_simplex_turn(monkeypatch): + from gateway.config import PlatformConfig + from gateway.platforms.base import MessageType, ProcessingOutcome + + cfg = PlatformConfig( + enabled=True, + extra={"ws_url": "ws://localhost:5225", "processing_notice_delay": 0.01}, + ) + adapter = SimplexAdapter(cfg) + sent = [] + + async def fake_send(chat_id, content, **_kwargs): + sent.append((chat_id, content)) + from gateway.platforms.base import SendResult + return SendResult(success=True) + + adapter.send = fake_send # type: ignore[method-assign] + source = adapter.build_source( + chat_id="4", + chat_name="Elkim", + chat_type="dm", + user_id="4", + user_name="Elkim", + ) + event = _simplex.MessageEvent(source=source, text="slow", message_type=MessageType.TEXT) + + await adapter.on_processing_start(event) + await _simplex.asyncio.sleep(0.03) + + assert sent == [("4", "Still here — SimpleX has no typing indicator, but I’m working on it.")] + await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS) def test_simplex_adapter_opts_into_active_session_hard_timeout(): @@ -419,6 +688,32 @@ async def fake_handle(wrapper): assert task not in adapter._poll_dispatch_tasks +@pytest.mark.asyncio +async def test_dispatch_polled_items_preserves_tail_order_for_bursts(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + handled = [] + + async def fake_handle(wrapper): + # If the burst were dispatched as independent tasks, the second item + # would finish first. Ordered batch dispatch must preserve daemon order. + if wrapper["item"] == 1: + await _simplex.asyncio.sleep(0.02) + handled.append(wrapper["item"]) + + adapter._handle_new_chat_item = fake_handle # type: ignore[method-assign] + task = adapter._dispatch_polled_items([ + ({"item": 1}, "direct:4:101"), + ({"item": 2}, "direct:4:102"), + ({"item": 3}, "direct:4:103"), + ]) + + await task + assert handled == [1, 2, 3] + assert task not in adapter._poll_dispatch_tasks + + @pytest.mark.asyncio async def test_stale_active_simplex_session_is_cancelled_for_fresh_message(): from gateway.config import PlatformConfig From 0efe08192f940ba6827c6bd3b2ca3da10a26884d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=BDofka?= Date: Sun, 17 May 2026 13:54:45 +0200 Subject: [PATCH 4/4] fix(simplex): isolate polling from gateway event loop --- plugins/platforms/simplex/adapter.py | 251 ++++++++++++++++++--------- tests/gateway/test_simplex_plugin.py | 40 +++++ 2 files changed, 212 insertions(+), 79 deletions(-) diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py index a75a3e1d77f41..6205f484b3b70 100644 --- a/plugins/platforms/simplex/adapter.py +++ b/plugins/platforms/simplex/adapter.py @@ -32,14 +32,16 @@ """ import asyncio +import concurrent.futures import json import logging import os import random +import threading import time from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union # Lazy import: BasePlatformAdapter and friends live in the main repo. # Imported at module top because they're stdlib-only inside Hermes — no @@ -148,6 +150,9 @@ def __init__(self, config: PlatformConfig, **kwargs): self._ws_task: Optional[asyncio.Task] = None self._health_task: Optional[asyncio.Task] = None self._poll_task: Optional[asyncio.Task] = None + self._poll_thread: Optional[threading.Thread] = None + self._poll_stop = threading.Event() + self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None self._typing_tasks: Dict[str, asyncio.Task] = {} self._running = False self._last_ws_activity = 0.0 @@ -155,6 +160,7 @@ def __init__(self, config: PlatformConfig, **kwargs): self._seen_items_path = get_hermes_home() / "simplex_seen_items.json" self._connected_at = 0.0 self._poll_dispatch_tasks: set[asyncio.Task] = set() + self._poll_dispatch_futures: set[concurrent.futures.Future] = set() self._read_receipt_tasks: set[asyncio.Task] = set() self._last_poll_started_at = 0.0 self._processing_notice_tasks: Dict[str, asyncio.Task] = {} @@ -210,13 +216,14 @@ async def connect(self) -> bool: return False self._running = True + self._gateway_loop = asyncio.get_running_loop() self._last_ws_activity = time.time() self._connected_at = time.time() self._load_seen_items() await self._seed_seen_items() self._ws_task = asyncio.create_task(self._ws_listener()) self._health_task = asyncio.create_task(self._health_monitor()) - self._poll_task = asyncio.create_task(self._poll_unread_items()) + self._start_poll_thread() logger.info("SimpleX: connected to %s", self.ws_url) return True @@ -224,6 +231,7 @@ async def connect(self) -> bool: async def disconnect(self) -> None: """Stop WebSocket listener and clean up.""" self._running = False + self._stop_poll_thread() if self._ws_task: self._ws_task.cancel() @@ -252,6 +260,10 @@ async def disconnect(self) -> None: await asyncio.gather(*self._poll_dispatch_tasks, return_exceptions=True) self._poll_dispatch_tasks.clear() + for future in list(self._poll_dispatch_futures): + future.cancel() + self._poll_dispatch_futures.clear() + for task in list(self._read_receipt_tasks): task.cancel() if self._read_receipt_tasks: @@ -277,6 +289,62 @@ async def disconnect(self) -> None: logger.info("SimpleX: disconnected") + def _start_poll_thread(self) -> None: + """Run the `/tail 50` fallback from a dedicated thread. + + Gateway message handling and agent/tool execution can monopolize the + main asyncio loop for long stretches. SimpleX polling is the recovery + path for missed daemon push events, so it must keep checking daemon + truth even while a Matrix turn or tool call is busy. + """ + if self._poll_thread and self._poll_thread.is_alive(): + return + self._poll_stop.clear() + self._poll_thread = threading.Thread( + target=self._poll_thread_main, + name="simplex-poll", + daemon=True, + ) + self._poll_thread.start() + + def _stop_poll_thread(self) -> None: + self._poll_stop.set() + thread = self._poll_thread + if thread and thread.is_alive(): + thread.join(timeout=max(POLL_WALL_TIMEOUT + 1.0, 5.0)) + if thread.is_alive(): + logger.warning("SimpleX poll: worker thread did not stop cleanly") + self._poll_thread = None + + def _poll_thread_main(self) -> None: + loop = asyncio.new_event_loop() + try: + asyncio.set_event_loop(loop) + loop.run_until_complete(self._poll_unread_items_thread()) + except Exception: + logger.exception("SimpleX poll: worker thread crashed") + finally: + try: + pending = [task for task in asyncio.all_tasks(loop) if not task.done()] + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + finally: + loop.close() + + async def _poll_unread_items_thread(self) -> None: + while self._running and not self._poll_stop.is_set(): + await asyncio.sleep(POLL_INTERVAL) + if not self._running or self._poll_stop.is_set(): + break + try: + await self._poll_unread_items_once() + except asyncio.CancelledError: + break + except Exception: + logger.exception("SimpleX poll: failed to poll unread items") + def _item_key(self, wrapper: dict) -> Optional[str]: """Return a stable per-chat key for a SimpleX chat item.""" chat_info = wrapper.get("chatInfo") or wrapper.get("chat") or {} @@ -584,88 +652,95 @@ async def _seed_seen_items(self) -> None: except Exception as e: logger.debug("SimpleX poll: initial seed failed: %s", e) - async def _poll_unread_items(self) -> None: - """Poll recent chat history for unread inbound items missed by push WS.""" - while self._running: - await asyncio.sleep(POLL_INTERVAL) - try: - start = time.time() - if self._last_poll_started_at: - poll_gap = start - self._last_poll_started_at - if poll_gap > POLL_STALL_WARN_SECONDS: - logger.warning( - "SimpleX poll: loop stalled for %.2fs before next /tail", - poll_gap, - ) - self._last_poll_started_at = start - resp = await self._command_once( - "/tail 50", - timeout=POLL_COMMAND_TIMEOUT, - open_timeout=POLL_CONNECT_TIMEOUT, - wall_timeout=POLL_WALL_TIMEOUT, - ) - poll_elapsed = time.time() - start - wrappers = (resp or {}).get("resp", {}).get("chatItems", []) or [] - if resp is None: - logger.warning( - "SimpleX poll: /tail 50 timed out after %.2fs", - poll_elapsed, - ) - elif poll_elapsed > POLL_COMMAND_TIMEOUT: - logger.warning( - "SimpleX poll: /tail 50 slow response %.2fs", - poll_elapsed, - ) - logger.debug( - "SimpleX poll: got %d items in %.2fs (seen=%d)", - len(wrappers), poll_elapsed, len(self._seen_item_ids) + async def _poll_unread_items_once(self) -> None: + start = time.time() + if self._last_poll_started_at: + poll_gap = start - self._last_poll_started_at + if poll_gap > POLL_STALL_WARN_SECONDS: + logger.warning( + "SimpleX poll: loop stalled for %.2fs before next /tail", + poll_gap, ) - dispatch_items: list[tuple[dict, str]] = [] - for wrapper in wrappers: - chat_item = wrapper.get("chatItem") or {} - meta = chat_item.get("meta") or {} - item_key = self._item_key(wrapper) - if item_key is None: - continue - if item_key in self._seen_item_ids: - continue - if not self._is_unread_inbound_text(wrapper): - item_ts = self._item_timestamp_epoch(wrapper) - if item_ts is not None and self._connected_at and item_ts < self._connected_at - 5: - self._seen_item_ids.add(item_key) - self._save_seen_items() - logger.info( - "SimpleX poll: marking stale pre-connect item seen: %s", - item_key, - ) - continue - status = (meta.get("itemStatus") or {}).get("type", "") - if status != "rcvNew": - self._seen_item_ids.add(item_key) - if len(self._seen_item_ids) > 1000: - self._save_seen_items() - continue - content = chat_item.get("content") or {} - if content.get("type") != "rcvMsgContent": - self._seen_item_ids.add(item_key) - if len(self._seen_item_ids) > 1000: - self._save_seen_items() - continue + self._last_poll_started_at = start + resp = await self._command_once( + "/tail 50", + timeout=POLL_COMMAND_TIMEOUT, + open_timeout=POLL_CONNECT_TIMEOUT, + wall_timeout=POLL_WALL_TIMEOUT, + ) + poll_elapsed = time.time() - start + wrappers = (resp or {}).get("resp", {}).get("chatItems", []) or [] + if resp is None: + logger.warning( + "SimpleX poll: /tail 50 timed out after %.2fs", + poll_elapsed, + ) + elif poll_elapsed > POLL_COMMAND_TIMEOUT: + logger.warning( + "SimpleX poll: /tail 50 slow response %.2fs", + poll_elapsed, + ) + logger.debug( + "SimpleX poll: got %d items in %.2fs (seen=%d)", + len(wrappers), poll_elapsed, len(self._seen_item_ids) + ) + dispatch_items: list[tuple[dict, str]] = [] + for wrapper in wrappers: + chat_item = wrapper.get("chatItem") or {} + meta = chat_item.get("meta") or {} + item_key = self._item_key(wrapper) + if item_key is None: + continue + if item_key in self._seen_item_ids: + continue + if not self._is_unread_inbound_text(wrapper): + item_ts = self._item_timestamp_epoch(wrapper) + if item_ts is not None and self._connected_at and item_ts < self._connected_at - 5: + self._seen_item_ids.add(item_key) + self._save_seen_items() logger.info( - "SimpleX poll: dispatching unread item %s", + "SimpleX poll: marking stale pre-connect item seen: %s", item_key, ) + continue + status = (meta.get("itemStatus") or {}).get("type", "") + if status != "rcvNew": self._seen_item_ids.add(item_key) - self._save_seen_items() - dispatch_items.append((wrapper, item_key)) - if dispatch_items: - self._dispatch_polled_items(dispatch_items) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() + continue + content = chat_item.get("content") or {} + if content.get("type") != "rcvMsgContent": + self._seen_item_ids.add(item_key) + if len(self._seen_item_ids) > 1000: + self._save_seen_items() + continue + logger.info( + "SimpleX poll: dispatching unread item %s", + item_key, + ) + self._seen_item_ids.add(item_key) + self._save_seen_items() + dispatch_items.append((wrapper, item_key)) + if dispatch_items: + self._dispatch_polled_items(dispatch_items) + + async def _poll_unread_items(self) -> None: + """Poll recent chat history for unread inbound items missed by push WS.""" + while self._running: + await asyncio.sleep(POLL_INTERVAL) + try: + await self._poll_unread_items_once() except asyncio.CancelledError: break except Exception: logger.exception("SimpleX poll: failed to poll unread items") - def _dispatch_polled_item(self, wrapper: dict, item_key: str) -> asyncio.Task: + def _dispatch_polled_item( + self, + wrapper: dict, + item_key: str, + ) -> Union[asyncio.Task, concurrent.futures.Future]: """Dispatch one polled item asynchronously. Kept as a small compatibility wrapper for tests and callers that hand us @@ -675,7 +750,10 @@ def _dispatch_polled_item(self, wrapper: dict, item_key: str) -> asyncio.Task: """ return self._dispatch_polled_items([(wrapper, item_key)]) - def _dispatch_polled_items(self, items: list[tuple[dict, str]]) -> asyncio.Task: + def _dispatch_polled_items( + self, + items: list[tuple[dict, str]], + ) -> Union[asyncio.Task, concurrent.futures.Future]: """Dispatch polled items in order while keeping the poll loop free.""" first_key = items[0][1] if items else "empty" started = time.time() @@ -710,10 +788,25 @@ async def _runner() -> None: batch_elapsed, ) - task = asyncio.create_task(_runner(), name=f"simplex-poll-dispatch:{first_key}") - self._poll_dispatch_tasks.add(task) - task.add_done_callback(self._poll_dispatch_tasks.discard) - return task + try: + running_loop = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + + if running_loop is not None: + task = running_loop.create_task(_runner(), name=f"simplex-poll-dispatch:{first_key}") + self._poll_dispatch_tasks.add(task) + task.add_done_callback(self._poll_dispatch_tasks.discard) + return task + + gateway_loop = self._gateway_loop + if gateway_loop is None or not gateway_loop.is_running(): + raise RuntimeError("SimpleX poll dispatch has no running gateway loop") + + future = asyncio.run_coroutine_threadsafe(_runner(), gateway_loop) + self._poll_dispatch_futures.add(future) + future.add_done_callback(self._poll_dispatch_futures.discard) + return future async def _resolve_chat_target(self, chat_id: str) -> str: """Resolve Hermes IDs to simplex-chat command targets. diff --git a/tests/gateway/test_simplex_plugin.py b/tests/gateway/test_simplex_plugin.py index c21e6d18d018a..70add200905a3 100644 --- a/tests/gateway/test_simplex_plugin.py +++ b/tests/gateway/test_simplex_plugin.py @@ -414,6 +414,46 @@ async def fake_command_once(_cmd, **_kwargs): assert "direct:4:99" in adapter._seen_item_ids +@pytest.mark.asyncio +async def test_connect_runs_polling_outside_gateway_event_loop(monkeypatch): + """SimpleX polling must not starve behind long gateway/agent turns.""" + import sys + import types + + from gateway.config import PlatformConfig + + class FakeConnect: + async def __aenter__(self): + return object() + + async def __aexit__(self, *_exc): + return False + + fake_websockets = types.SimpleNamespace(connect=lambda *_a, **_kw: FakeConnect()) + monkeypatch.setitem(sys.modules, "websockets", fake_websockets) + + cfg = PlatformConfig(enabled=True, extra={"ws_url": "ws://localhost:5225"}) + adapter = SimplexAdapter(cfg) + + async def noop_seed(): + return None + + async def idle_task(): + await _simplex.asyncio.sleep(60) + + adapter._seed_seen_items = noop_seed # type: ignore[method-assign] + adapter._ws_listener = idle_task # type: ignore[method-assign] + adapter._health_monitor = idle_task # type: ignore[method-assign] + + assert await adapter.connect() is True + try: + assert adapter._poll_task is None + assert adapter._poll_thread is not None + assert adapter._poll_thread.is_alive() + finally: + await adapter.disconnect() + + @pytest.mark.asyncio async def test_poll_dispatches_preconnect_unread_inbound_text(monkeypatch, tmp_path): """Unread user text remains actionable even if its timestamp predates reconnect."""