ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(gateway): adapters share helpers.cancel_task / MessageDeduplicator / bounded_put

Browse Source 
Six adapters defined their own `_cancel_task` and nine more inlined the same
cancel + suppress(CancelledError) + await block; five kept a hand-rolled TTL-dict
`_is_duplicate` next to the existing `helpers.MessageDeduplicator`; three carried a
`_bounded_put`. Each copy fixed the same bugs on its own schedule (self-cancel deadlock,
done-task re-await, eviction under load).

- `helpers.cancel_task`: None/done no-op, never awaits the current task, swallows the
  task's own exception at teardown. Replaces qqbot/signal/yuanbao/buzz/photon/simplex
  definitions and the inline copies in weixin, discord, email, irc, line, mattermost,
  whatsapp and telegram.
- `helpers.MessageDeduplicator` replaces `_is_duplicate` in qqbot, ntfy, photon,
  wecom_callback and LINE's `_MessageDeduplicator`; every site keeps its own
  max_size/TTL (qqbot and ntfy 1000/300s, photon 4000/48h, wecom_callback 2000/300s,
  LINE 1000/no TTL).
- `helpers.bounded_put` replaces photon/wecom/whatsapp_cloud copies; a re-put now
  refreshes the key to the newest slot at every site.
- telegram gmail-triage scripts resolve under `get_hermes_home()` instead of a hard
  `~/.hermes`, so profiles with HERMES_HOME set find them.

Not changed: `get_chat_info` stays `@abstractmethod` because
tests/gateway/test_relay_capability_surface.py locks the abstract set to exactly
{connect, disconnect, send, get_chat_info} as a cross-repo contract, so the ~17 no-op
overrides remain.

Behavior change: whatsapp_cloud `_bounded_put` was a pure FIFO (no refresh on re-put);
it now refreshes like the other two sites. Task cancellation at the migrated sites
swallows a task's terminal exception where a few copies previously only suppressed
CancelledError (all are shutdown/disconnect paths).
This commit is contained in:
teknium1
2026-09-12 20:07:31 -07:00
committed by Teknium
parent 3a179fe524
commit 9b1990583d
24 changed files with 198 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/helpers.py
+24
View File
@@ -2,11 +2,14 @@
stripping, thread participation tracking, GFM table → bullets, mention-pattern
compilation, and fence-aware markdown chunking."""
import asyncio
import contextlib
import json
import logging
import re
import time
from pathlib import Path
from typing import Any, MutableMapping, Optional
from gateway.platforms.event import MessageEvent
from utils import atomic_json_write
@@ -57,6 +60,27 @@ class MessageDeduplicator:
self._seen.clear()
async def cancel_task(task: Optional[asyncio.Task]) -> None:
"""Cancel *task* and wait for it to unwind. ``None``/finished tasks are no-ops; awaiting the
current task would deadlock, so a self-cancel only requests cancellation. Exceptions the task
dies with are swallowed: at teardown nobody is left to handle them."""
if task is None or task.done():
return
task.cancel()
if task is not asyncio.current_task():
with contextlib.suppress(asyncio.CancelledError, Exception):
await task
def bounded_put(store: MutableMapping[str, Any], key: str, value: Any, cap: int) -> None:
"""Insert into an insertion-ordered mapping with a hard size bound, evicting the oldest keys. A
re-put moves the key to the newest position so live entries outlast stale ones."""
store.pop(key, None)
store[key] = value
while len(store) > cap:
del store[next(iter(store))]
# Markdown-stripping rules, applied in order: bold, italic, bold/italic underscore,
# code fence markers, inline code, headings, links, then newline squeeze.
_STRIP_RULES = (
gateway/platforms/qqbot/adapter.py
+6 -23
View File
@@ -44,6 +44,7 @@ from gateway.platforms.base import (
)
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms.helpers import strip_markdown
from gateway.platforms.helpers import MessageDeduplicator, cancel_task
from gateway.platforms.access_policy_mixin import OwnAccessPolicyMixin
from gateway.platforms.media_cache import ext_for_mime
@@ -162,7 +163,7 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
self._last_seq: Optional[int] = None
self._chat_type_map: Dict[str, str] = {} # chat_id → "c2c"|"group"|"guild"|"dm"
self._pending_responses: Dict[str, asyncio.Future] = {} # request/response correlation
self._seen_messages: Dict[str, float] = {}
self._dedup = MessageDeduplicator(max_size=DEDUP_MAX_SIZE, ttl_seconds=DEDUP_WINDOW_SECONDS)
self._last_msg_id: Dict[str, str] = {} # last inbound message ID per chat (send_typing)
self._typing_sent_at: Dict[str, float] = {} # typing debounce: chat_id → last send_typing ts
self._access_token: Optional[str] = None
@@ -227,21 +228,13 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
async def disconnect(self) -> None:
self._running = False
self._mark_disconnected()
self._listen_task = await self._cancel_task(self._listen_task)
self._heartbeat_task = await self._cancel_task(self._heartbeat_task)
await cancel_task(self._listen_task)
await cancel_task(self._heartbeat_task)
self._listen_task = self._heartbeat_task = None
await self._cleanup()
self._release_platform_lock()
logger.info("[%s] Disconnected", self._log_tag)
@staticmethod
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
"""Cancel and await *task* (if any); always returns None for reassignment."""
if task:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
return None
async def _close_ws(self) -> None:
"""Close the WebSocket + its aiohttp session (keeps _http_client alive)."""
if self._ws and not self._ws.closed:
@@ -585,7 +578,7 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
if not isinstance(d, dict):
return
msg_id = str(d.get("id", ""))
if not msg_id or self._is_duplicate(msg_id):
if not msg_id or self._dedup.is_duplicate(msg_id):
logger.debug("[%s] Duplicate or missing message id: %s", self._log_tag, msg_id)
return
handler = self._INBOUND_HANDLERS.get(event_type)
@@ -1670,16 +1663,6 @@ class QQAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc)
return datetime.now(tz=timezone.utc)
def _is_duplicate(self, msg_id: str) -> bool:
now = time.time()
if len(self._seen_messages) > DEDUP_MAX_SIZE:
cutoff = now - DEDUP_WINDOW_SECONDS
self._seen_messages = {k: ts for k, ts in self._seen_messages.items() if ts > cutoff}
if msg_id in self._seen_messages:
return True
self._seen_messages[msg_id] = now
return False
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
gateway/platforms/signal.py
+3 -9
View File
@@ -30,6 +30,7 @@ from gateway.platforms.base import (
)
from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome
from gateway.platforms.helpers import redact_phone
from gateway.platforms.helpers import cancel_task
from gateway.platforms.media_cache import mime_for_ext
from tools.audio_container import CONTAINER_TO_EXT, sniff_container
from gateway.platforms.signal_format import markdown_to_signal
@@ -265,18 +266,11 @@ class SignalAdapter(BasePlatformAdapter):
await self.client.aclose()
self.client = None
@staticmethod
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
if task:
task.cancel()
with suppress(asyncio.CancelledError):
await task
async def disconnect(self) -> None:
"""Stop SSE listener and clean up."""
self._running = False
for task in (self._sse_task, self._health_monitor_task):
await self._cancel_task(task)
await cancel_task(task)
for task in self._typing_tasks.values():
task.cancel()
self._typing_tasks.clear()
@@ -914,7 +908,7 @@ class SignalAdapter(BasePlatformAdapter):
async def _stop_typing_indicator(self, chat_id: str) -> None:
"""Stop a typing indicator loop for a chat."""
await self._cancel_task(self._typing_tasks.pop(chat_id, None))
await cancel_task(self._typing_tasks.pop(chat_id, None))
# Explicit stop-typing RPC so the recipient drops the indicator now instead of after
# Signal's ~5s timeout. Best-effort: failures must not prevent the backoff cleanup below.
with suppress(Exception):
gateway/platforms/weixin.py
+2 -5
View File
@@ -28,7 +28,7 @@ except ImportError: # pragma: no cover - dependency gate
CRYPTO_AVAILABLE = Cipher is not None
from gateway.config import Platform, PlatformConfig
from gateway.platforms.helpers import MessageDeduplicator, greedy_pack_blocks
from gateway.platforms.helpers import MessageDeduplicator, cancel_task, greedy_pack_blocks
from gateway.platforms.access_policy_mixin import OwnAccessPolicyMixin
from gateway.platforms.base import (
_IMAGE_EXTS, _VIDEO_EXTS, gateway_trust_env, BasePlatformAdapter, SendResult,
@@ -797,10 +797,7 @@ class WeixinAdapter(OwnAccessPolicyMixin, BasePlatformAdapter):
task.cancel()
self._pending_text_batches.clear()
self._pending_text_batch_tasks.clear()
if self._poll_task and not self._poll_task.done():
self._poll_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._poll_task
await cancel_task(self._poll_task)
self._poll_task = None
for attr in ("_poll_session", "_send_session"):
session = getattr(self, attr)
gateway/platforms/whatsapp_cloud.py
+5 -11
View File
@@ -40,6 +40,7 @@ except ImportError:
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, ExecApprovalPrompt, SendResult, transcode_to_ogg_opus
from gateway.platforms.helpers import bounded_put
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms.whatsapp_common import WhatsAppBehaviorMixin, _get_wsecret
from gateway.platforms.access_policy_mixin import OPTIN_TRUTHY as _OPTIN_TRUTHY
@@ -203,7 +204,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
# message_id to attach to, and the base send_typing contract has none.
self._last_inbound_wamid_by_chat: "OrderedDict[str, str]" = OrderedDict()
# Interactive-button state: short id (in the button payload) → session_key for
# the gateway resolver. Popped on tap; FIFO-capped via _bounded_put so ignored
# the gateway resolver. Popped on tap; FIFO-capped via bounded_put so ignored
# prompts don't accumulate (an evicted tap degrades to text fallback).
self._clarify_state: "OrderedDict[str, str]" = OrderedDict()
self._exec_approval_state: "OrderedDict[str, str]" = OrderedDict()
@@ -224,13 +225,6 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
headers = {"Authorization": f"Bearer {self._access_token}"}
return {**headers, "Content-Type": "application/json"} if json_body else headers
@staticmethod
def _bounded_put(cache: "OrderedDict[str, Any]", key: str, value: Any, cap: int = INTERACTIVE_STATE_CACHE_SIZE) -> None:
"""Insert into a FIFO-capped OrderedDict, evicting oldest entries."""
cache[key] = value
while len(cache) > cap:
cache.popitem(last=False)
def _effective_reply_prefix(self) -> str:
"""Cloud API has no self-chat concept (a Baileys-only setting) — no default prefix."""
return self._reply_prefix.replace("\\n", "\n") if self._reply_prefix is not None else ""
@@ -416,7 +410,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
reject_log="[whatsapp_cloud] interactive rejected (status=%d): %s",
)
if result.success:
self._bounded_put(state, state_id, session_key)
bounded_put(state, state_id, session_key, INTERACTIVE_STATE_CACHE_SIZE)
return result
@staticmethod
@@ -737,7 +731,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
self._duplicate_count += 1
return False
if wamid: # can't dedup without an id — let it through
self._bounded_put(self._seen_wamids, wamid, True, cap=WAMID_DEDUP_CACHE_SIZE)
bounded_put(self._seen_wamids, wamid, True, WAMID_DEDUP_CACHE_SIZE)
return True
async def _dispatch_payload(self, payload: Dict[str, Any]) -> None:
@@ -992,7 +986,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
wamid = str(raw_message.get("id") or "") or None
if wamid and chat_id:
# Done AFTER gating so filtered messages don't leak typing/read receipts.
self._bounded_put(self._last_inbound_wamid_by_chat, chat_id, wamid)
bounded_put(self._last_inbound_wamid_by_chat, chat_id, wamid, INTERACTIVE_STATE_CACHE_SIZE)
if body:
rich_sent_store.record(chat_id, wamid, body)
if msg_type_str in _INBOUND_MEDIA_KINDS and media_urls:
gateway/platforms/yuanbao.py
+3 -10
View File
@@ -49,7 +49,7 @@ from gateway.platforms.base import (
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms import helpers as _mdchunk
from gateway.platforms._shared import get_scoped_secret as _yb_secret, profile_scoped as _profile_scoped
from gateway.platforms.helpers import MessageDeduplicator
from gateway.platforms.helpers import MessageDeduplicator, cancel_task
from gateway.platforms.access_policy_mixin import OwnAccessPolicyMixin
from gateway.platforms.yuanbao_media import (
download_url as media_download_url, get_cos_credentials, upload_to_cos,
@@ -139,13 +139,6 @@ def _cancel_all(tasks: Dict[str, asyncio.Task]) -> None:
tasks.clear()
async def _cancel_task(task: asyncio.Task) -> None:
"""Cancel *task* and wait for it to unwind (swallowing the CancelledError)."""
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
class MarkdownProcessor:
"""Yuanbao's fence/table-aware chunking policy over the shared chunker in gateway.platforms.helpers."""
@classmethod
@@ -1869,7 +1862,7 @@ class ConnectionManager:
for attr, _coro_name, _tag in self._LOOPS:
task = getattr(self, attr)
if task:
await _cancel_task(task)
await cancel_task(task)
setattr(self, attr, None)
disc_exc = RuntimeError("YuanbaoAdapter disconnected")
for fut in self._pending_acks.values():
@@ -2336,7 +2329,7 @@ class HeartbeatManager:
"""Stop the RUNNING sender and optionally send FINISH."""
task = self._reply_heartbeat_tasks.pop(chat_id, None)
if task and not task.done():
await _cancel_task(task)
await cancel_task(task)
if send_finish:
await self.send_heartbeat_once(chat_id, WS_HEARTBEAT_FINISH)
plugins/platforms/buzz/adapter.py
+4 -12
View File
@@ -194,6 +194,7 @@ logger = logging.getLogger(__name__)
from gateway.platforms.base import (
BasePlatformAdapter, CachedMedia, SendResult, cache_media_bytes_async,
)
from gateway.platforms.helpers import cancel_task
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform
@@ -820,22 +821,13 @@ class BuzzAdapter(BasePlatformAdapter):
self._mark_disconnected()
with contextlib.suppress(Exception):
self._release_platform_lock()
await self._cancel_task(self._ws_task)
await cancel_task(self._ws_task)
self._ws_task = None
await self._cancel_task(self._poll_task)
await cancel_task(self._poll_task)
self._poll_task = None
self._channel_state = {}
self._poll_count = 0
@staticmethod
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
if task and not task.done():
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
# ── Sending ───────────────────────────────────────────────────────────
async def _channel_member_pubkeys(self, chat_id: str) -> List[str]:
@@ -1116,7 +1108,7 @@ class BuzzAdapter(BasePlatformAdapter):
return True
except (asyncio.TimeoutError, TimeoutError):
logger.warning("Buzz: WebSocket did not authenticate in time")
await self._cancel_task(self._ws_task)
await cancel_task(self._ws_task)
self._ws_task = None
return False
plugins/platforms/discord/adapter.py
+2 -4
View File
@@ -260,6 +260,7 @@ from gateway.config import Platform, PlatformConfig
from gateway.platforms.helpers import (
MessageDeduplicator, ThreadParticipationTracker, convert_table_to_bullets,
)
from gateway.platforms.helpers import cancel_task
from utils import atomic_json_write, env_float
from gateway.platforms.base import (
BasePlatformAdapter, ExecApprovalPrompt, SendResult,
@@ -343,10 +344,7 @@ async def _wait_for_ready_or_bot_exit(
raise RuntimeError("Discord bot task exited before ready")
await ready_task
finally:
if not ready_task.done():
ready_task.cancel()
with suppress(asyncio.CancelledError):
await ready_task
await cancel_task(ready_task)
def _needs_server_members_intent(
plugins/platforms/email/adapter.py
+3 -5
View File
@@ -25,6 +25,7 @@ from gateway.platforms.base import (
BasePlatformAdapter, SendResult,
cache_document_from_bytes, cache_image_from_bytes,
)
from gateway.platforms.helpers import cancel_task
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform, PlatformConfig
from utils import is_truthy_value
@@ -492,11 +493,8 @@ class EmailAdapter(BasePlatformAdapter):
async def disconnect(self) -> None:
"""Stop polling and disconnect."""
self._running = False
if self._poll_task:
self._poll_task.cancel()
with suppress(asyncio.CancelledError):
await self._poll_task
self._poll_task = None
await cancel_task(self._poll_task)
self._poll_task = None
logger.info("[Email] Disconnected.")
async def _poll_loop(self) -> None:
plugins/platforms/irc/adapter.py
+2 -4
View File
@@ -17,6 +17,7 @@ from typing import Any, Dict, List, Optional
from gateway.platforms._shared import coerce_port, get_scoped_secret as _get_scoped_secret, send_error
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.helpers import cancel_task
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform
@@ -198,10 +199,7 @@ class IRCAdapter(BasePlatformAdapter):
with contextlib.suppress(Exception):
self._writer.close()
await self._writer.wait_closed()
if self._recv_task and not self._recv_task.done():
self._recv_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._recv_task
await cancel_task(self._recv_task)
self._reader = None
self._writer = None
self._registered = False
plugins/platforms/line/adapter.py
+4 -24
View File
@@ -41,6 +41,7 @@ from gateway.platforms.base import (
cache_audio_from_bytes_async, cache_document_from_bytes_async, cache_image_from_bytes_async,
cache_video_from_bytes_async,
)
from gateway.platforms.helpers import MessageDeduplicator, cancel_task
from gateway.platforms.event import MessageEvent, MessageType
from gateway.config import Platform
@@ -190,25 +191,6 @@ class RequestCache:
self._transition(request_id, {State.READY, State.ERROR}, State.DELIVERED)
class _MessageDeduplicator:
"""Bounded LRU of LINE webhook event IDs to ignore at-least-once retries."""
def __init__(self, max_size: int = 1000) -> None:
self._seen: Dict[str, float] = {}
self._max = max_size
def is_duplicate(self, event_id: str) -> bool:
if not event_id:
return False
if event_id in self._seen:
return True
if len(self._seen) >= self._max: # drop the oldest 10% so we don't trim every insert
cutoff = sorted(self._seen.values())[len(self._seen) // 10 or 1]
self._seen = {k: v for k, v in self._seen.items() if v > cutoff}
self._seen[event_id] = time.time()
return False
# LINE source type → (id key, normalized chat_type)
_SOURCE_KINDS = {"group": ("groupId", "group"), "room": ("roomId", "room"), "user": ("userId", "dm")}
@@ -423,7 +405,8 @@ class LineAdapter(BasePlatformAdapter):
self._app = self._runner = self._site = None # aiohttp web.Application / AppRunner / TCPSite
self._reply_tokens: Dict[str, Tuple[str, float]] = {} # chat_id → (token, expiry)
self._cache = RequestCache()
self._dedup = _MessageDeduplicator()
# LINE redelivers webhooks for up to a day on non-2xx; no TTL, just a size bound.
self._dedup = MessageDeduplicator(max_size=1000, ttl_seconds=float("inf"))
self._bot_user_id: Optional[str] = None
self._media_tokens: Dict[str, Tuple[str, float]] = {} # token → (path, expiry)
self._media_temp_paths: Set[str] = set()
@@ -728,10 +711,7 @@ class LineAdapter(BasePlatformAdapter):
try:
await super()._keep_typing(chat_id, *args, **kwargs)
finally:
if not post_task.done():
post_task.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await post_task
await cancel_task(post_task)
async def interrupt_session_activity(self, session_key: str, chat_id: str) -> None:
"""Resolve any orphan PENDING postback so the button doesn't loop."""
plugins/platforms/mattermost/adapter.py
+2 -4
View File
@@ -22,6 +22,7 @@ from typing import Any, Dict, List, Optional, Tuple
from gateway.config import Platform, PlatformConfig
from gateway.platforms.helpers import MessageDeduplicator
from gateway.platforms.helpers import cancel_task
from gateway.platforms.base import gateway_trust_env, BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret, profile_scoped as _profile_scoped_config_load, send_error
@@ -251,10 +252,7 @@ class MattermostAdapter(BasePlatformAdapter):
async def disconnect(self) -> None:
self._closing = True
if self._ws_task and not self._ws_task.done():
self._ws_task.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await self._ws_task
await cancel_task(self._ws_task)
if self._reconnect_task and not self._reconnect_task.done():
self._reconnect_task.cancel()
if self._ws:
plugins/platforms/ntfy/adapter.py
+4 -14
View File
@@ -28,6 +28,7 @@ from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret, send_error
from gateway.platforms.helpers import MessageDeduplicator
logger = logging.getLogger(__name__)
@@ -129,7 +130,7 @@ class NtfyAdapter(BasePlatformAdapter):
self._token: str = _setting(extra, "token", "NTFY_TOKEN")
self._stream_task: Optional[asyncio.Task] = None
self._http_client: Optional["httpx.AsyncClient"] = None
self._seen_messages: Dict[str, float] = {} # msg_id -> timestamp (dedup)
self._dedup = MessageDeduplicator(max_size=DEDUP_MAX_SIZE, ttl_seconds=DEDUP_WINDOW_SECONDS)
# -- Connection lifecycle -----------------------------------------------
@@ -234,7 +235,7 @@ class NtfyAdapter(BasePlatformAdapter):
if self._http_client:
await self._http_client.aclose()
self._http_client = None
self._seen_messages.clear()
self._dedup.clear()
logger.info("[%s] Disconnected", self.name)
# -- Inbound message processing -----------------------------------------
@@ -242,7 +243,7 @@ class NtfyAdapter(BasePlatformAdapter):
async def _on_message(self, event: Dict[str, Any]) -> None:
"""Process an incoming ntfy message event."""
msg_id = event.get("id") or uuid.uuid4().hex
if self._is_duplicate(msg_id):
if self._dedup.is_duplicate(msg_id):
logger.debug("[%s] Duplicate message %s, skipping", self.name, msg_id)
return
if _ECHO_TAG in (event.get("tags") or []):
@@ -268,17 +269,6 @@ class NtfyAdapter(BasePlatformAdapter):
logger.debug("[%s] Message on topic %s: %s", self.name, topic, text[:80])
await self.handle_message(message_event)
def _is_duplicate(self, msg_id: str) -> bool:
"""True if this message ID was already seen within the dedup window."""
now = time.time()
if len(self._seen_messages) > DEDUP_MAX_SIZE:
cutoff = now - DEDUP_WINDOW_SECONDS
self._seen_messages = {k: v for k, v in self._seen_messages.items() if v > cutoff}
if msg_id in self._seen_messages:
return True
self._seen_messages[msg_id] = now
return False
# -- Outbound messaging -------------------------------------------------
async def send(
plugins/platforms/photon/adapter.py
+8 -33
View File
@@ -41,6 +41,7 @@ from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret, s
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms.helpers import compile_mention_patterns, strip_markdown
from gateway.platforms.helpers import MessageDeduplicator, bounded_put, cancel_task
from utils import atomic_json_write
from .auth import load_project_credentials
@@ -438,24 +439,6 @@ def _guess_mime(path: str) -> Optional[str]:
return mimetypes.guess_type(path)[0] or None
def _bounded_put(store: Dict[str, Any], key: str, value: Any, max_size: int) -> None:
"""Insert with insertion-order refresh and a HARD size bound (evict oldest)."""
if key in store:
del store[key]
store[key] = value
if len(store) > max_size:
for old in list(store.keys())[: len(store) - max_size]:
del store[old]
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
"""Cancel *task* and wait for it, unless it is the current task."""
if task is None:
return
task.cancel()
if task is not asyncio.current_task():
with contextlib.suppress(asyncio.CancelledError, Exception):
await task
# -- Adapter -------------------------------------------------------------------
@@ -506,7 +489,7 @@ class PhotonAdapter(BasePlatformAdapter):
self._sidecar_health_interval = 15.0
self._probe_failures = 0
self._last_upstream_activity = 0.0 # monotonic; watchdog skips probe if traffic proved liveness
self._seen_messages: Dict[str, float] = {} # at-least-once stream dedup
self._dedup = MessageDeduplicator(max_size=_DEDUP_MAX_SIZE, ttl_seconds=_DEDUP_WINDOW_SECONDS) # at-least-once stream
self._sent_message_ids: Dict[str, float] = {} # only reactions targeting OUR sends are routed
self._last_inbound_by_chat: Dict[str, str] = {} # default target for the react action
self._recent_richlinks_by_chat: Dict[str, float] = {} # coalesce preview-art attachments
@@ -596,9 +579,9 @@ class PhotonAdapter(BasePlatformAdapter):
self._inbound_running = False
await self._stop_watchdog() # first, so it can't respawn while we tear the sidecar down
task, self._sidecar_health_task = self._sidecar_health_task, None
await _cancel_task(task)
await cancel_task(task)
task, self._inbound_task = self._inbound_task, None
await _cancel_task(task)
await cancel_task(task)
for _, fffc_task in list(self._pending_fffc.values()):
if fffc_task and not fffc_task.done():
fffc_task.cancel()
@@ -698,21 +681,13 @@ class PhotonAdapter(BasePlatformAdapter):
logger.debug("[photon] skipping non-JSON inbound line")
return
msg_id = event.get("messageId")
if msg_id and self._is_duplicate(msg_id):
if msg_id and self._dedup.is_duplicate(msg_id):
return
try:
await self._dispatch_inbound(event)
except Exception:
logger.exception("[photon] inbound dispatch failed")
def _is_duplicate(self, msg_id: str) -> bool:
now = time.time()
t = self._seen_messages.get(msg_id)
if t is not None and now - t < _DEDUP_WINDOW_SECONDS:
return True
_bounded_put(self._seen_messages, msg_id, now, _DEDUP_MAX_SIZE)
return False
async def _fffc_timeout_handler(self, chat_key: str, message_id: str) -> None:
await asyncio.sleep(_FFFC_WAIT_SECONDS)
if self._pending_fffc.pop(chat_key, None):
@@ -1115,7 +1090,7 @@ class PhotonAdapter(BasePlatformAdapter):
async def _stop_watchdog(self) -> None:
self._watchdog_running = False
task, self._watchdog_task = self._watchdog_task, None
await _cancel_task(task)
await cancel_task(task)
# -- Outbound ------------------------------------------------------------------
@@ -1202,7 +1177,7 @@ class PhotonAdapter(BasePlatformAdapter):
def _record_sent_message(self, message_id: Optional[str]) -> None:
if message_id:
_bounded_put(self._sent_message_ids, message_id, time.time(), self._SENT_IDS_MAX)
bounded_put(self._sent_message_ids, message_id, time.time(), self._SENT_IDS_MAX)
# A DM space is addressable as the chat GUID (`any;-;+1555...`) inbound events carry, or
# the bare E.164 phone home-channel config uses; the sidecar's resolveSpace treats them
@@ -1215,7 +1190,7 @@ class PhotonAdapter(BasePlatformAdapter):
return match.group(1) if match else chat_id
def _put_by_chat(self, store: Dict[str, Any], chat_id: str, value: Any) -> None:
_bounded_put(store, self._normalize_chat_key(chat_id), value, self._LAST_INBOUND_CHATS_MAX)
bounded_put(store, self._normalize_chat_key(chat_id), value, self._LAST_INBOUND_CHATS_MAX)
def _record_last_inbound(self, chat_id: Optional[str], message_id: Optional[str]) -> None:
if chat_id and message_id:
plugins/platforms/simplex/adapter.py
+3 -9
View File
@@ -27,6 +27,7 @@ from urllib.parse import unquote
from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret, send_error
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult, cache_image_from_url
from gateway.platforms.helpers import cancel_task
from gateway.platforms.event import MessageEvent, MessageType
logger = logging.getLogger(__name__)
@@ -84,13 +85,6 @@ def _send_cmd(chat_id: str, items: list) -> str:
return f"/_send {target} json {json.dumps(items)}"
async def _cancel_task(task: Optional[asyncio.Task]) -> None:
if task:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
class SimplexAdapter(BasePlatformAdapter):
"""SimpleX Chat adapter using the simplex-chat daemon WebSocket API."""
@@ -155,8 +149,8 @@ class SimplexAdapter(BasePlatformAdapter):
async def disconnect(self) -> None:
self._running = False
await _cancel_task(self._ws_task)
await _cancel_task(self._health_task)
await cancel_task(self._ws_task)
await cancel_task(self._health_task)
if self._ws:
with contextlib.suppress(Exception):
await self._ws.close()
plugins/platforms/telegram/adapter.py
+3 -4
View File
@@ -288,6 +288,7 @@ def _separate_chunk_indicator_from_fence(text: str) -> str:
# MarkdownV2 has no table syntax, so pipe tables become bullet groups via convert_table_to_bullets().
from gateway.platforms.helpers import (
TABLE_SEPARATOR_RE as _TABLE_SEPARATOR_RE, compile_mention_patterns, convert_table_to_bullets as _wrap_markdown_tables)
from gateway.platforms.helpers import cancel_task
# Rich-message regions whose internal newlines must stay bare (Telegram renders them natively):
# fenced code blocks OR GFM pipe-table blocks (header row, delimiter row, data rows).
@@ -657,10 +658,8 @@ class TelegramAdapter(BasePlatformAdapter):
async def _redispatch_held_inbound(self, prior: Optional[asyncio.Task] = None) -> None:
"""Drain the hold queue after reconnect or a connected-path hold; ``prior`` (previous
redispatch task) is cancelled+awaited here so ``_mark_connected`` stays synchronous."""
if prior is not None and prior is not asyncio.current_task() and not prior.done():
prior.cancel()
with contextlib.suppress(asyncio.CancelledError):
await prior
if prior is not asyncio.current_task(): # a self-redispatch must not cancel itself
await cancel_task(prior)
held = getattr(self, "_held_inbound_events", None)
if self._is_permanent_fatal():
if held:
plugins/platforms/wecom/adapter.py
+7 -16
View File
@@ -27,7 +27,7 @@ AIOHTTP_AVAILABLE = aiohttp is not None
HTTPX_AVAILABLE = httpx is not None
from gateway.config import Platform, PlatformConfig
from gateway.platforms.helpers import MessageDeduplicator
from gateway.platforms.helpers import MessageDeduplicator, bounded_put
from gateway.platforms.access_policy_mixin import OwnAccessPolicyMixin
from gateway.platforms.base import gateway_trust_env, BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
@@ -98,18 +98,6 @@ def _content_of(container: Dict[str, Any], key: str) -> str:
return str(_dict_or_empty(container, key).get("content") or "").strip()
def _bounded_put(store: Dict[str, str], key: str, value: str) -> bool:
"""Insert into an insertion-ordered dict bounded at DEDUP_MAX_SIZE; False if key/value empty."""
key = str(key or "").strip()
value = str(value or "").strip()
if not key or not value:
return False
store[key] = value
while len(store) > DEDUP_MAX_SIZE:
store.pop(next(iter(store)))
return True
class WeComAdapter(WeComStreamMixin, WeComMediaMixin, ChatSendQueueMixin, OwnAccessPolicyMixin, BasePlatformAdapter):
"""WeCom AI Bot adapter backed by a persistent WebSocket connection."""
@@ -439,7 +427,8 @@ class WeComAdapter(WeComStreamMixin, WeComMediaMixin, ChatSendQueueMixin, OwnAcc
# INFO: a msgid redelivered after a processing exception is dropped for the TTL.
logger.info("[%s] Duplicate message %s ignored (dedup drop) req_id=%s sender=%r chattype=%r", self.name, msg_id, req_id, sender.get("userid") if sender else None, body.get("chattype"))
return
_bounded_put(self._reply_req_ids, msg_id, req_id)
if req_id:
bounded_put(self._reply_req_ids, msg_id, req_id, DEDUP_MAX_SIZE)
chat_id = str(body.get("chatid") or sender_id).strip()
logger.info("[%s] Inbound callback: chattype=%r chatid=%r sender=%r msgtype=%r has_chatid=%s", self.name, body.get("chattype"), body.get("chatid"), sender_id, body.get("msgtype"), bool(body.get("chatid")))
if not chat_id:
@@ -554,8 +543,10 @@ class WeComAdapter(WeComStreamMixin, WeComMediaMixin, ChatSendQueueMixin, OwnAcc
def _remember_chat_req_id(self, chat_id: str, req_id: str) -> None:
"""Cache the chat's latest inbound req_id; a fresh one also resurrects its stream channel."""
if _bounded_put(self._last_chat_req_ids, chat_id, req_id):
self._stream_expired_chats.discard(str(chat_id).strip())
chat_id, req_id = str(chat_id or "").strip(), str(req_id or "").strip()
if chat_id and req_id:
bounded_put(self._last_chat_req_ids, chat_id, req_id, DEDUP_MAX_SIZE)
self._stream_expired_chats.discard(chat_id)
def _reply_req_id_for_message(self, reply_to: Optional[str]) -> Optional[str]:
normalized = str(reply_to or "").strip()
plugins/platforms/wecom/callback_adapter.py
+3 -13
View File
@@ -34,6 +34,7 @@ except ImportError:
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.platforms.event import MessageEvent, MessageType
from gateway.platforms.helpers import MessageDeduplicator
from plugins.platforms.wecom.wecom_crypto import WXBizMsgCrypt, WeComCryptoError
logger = logging.getLogger(__name__)
@@ -93,7 +94,7 @@ class WecomCallbackAdapter(BasePlatformAdapter):
self._apps: List[Dict[str, Any]] = self._normalize_apps(extra)
self._runner = self._site = self._app = self._http_client = self._poll_task = None
self._message_queue: asyncio.Queue[MessageEvent] = asyncio.Queue()
self._seen_messages: Dict[str, float] = {}
self._dedup = MessageDeduplicator(ttl_seconds=MESSAGE_DEDUP_TTL_SECONDS)
self._user_app_map: Dict[str, str] = {}
self._access_tokens: Dict[str, Dict[str, Any]] = {}
@@ -233,7 +234,7 @@ class WecomCallbackAdapter(BasePlatformAdapter):
event = self._build_event(app, self._decrypt_request(app, body, msg_signature, timestamp, nonce))
if event is not None:
# WeCom retries callbacks on timeout → duplicate inbound messages.
if event.message_id and self._is_duplicate(event.message_id):
if event.message_id and self._dedup.is_duplicate(event.message_id):
logger.debug("[WecomCallback] Duplicate MsgId %s, skipping", event.message_id)
return _ack()
if event.source and event.source.user_id:
@@ -251,17 +252,6 @@ class WecomCallbackAdapter(BasePlatformAdapter):
def _signature_params(request: web.Request):
return tuple(request.query.get(k, "") for k in ("msg_signature", "timestamp", "nonce"))
def _is_duplicate(self, message_id: str) -> bool:
# Deduplicate: WeCom retries callbacks on timeout, producing duplicate inbound messages (#10305).
now = time.time()
if now - self._seen_messages.get(message_id, float("-inf")) < MESSAGE_DEDUP_TTL_SECONDS:
return True
self._seen_messages[message_id] = now
if len(self._seen_messages) > 2000: # prune expired entries
cutoff = now - MESSAGE_DEDUP_TTL_SECONDS
self._seen_messages = {k: v for k, v in self._seen_messages.items() if v > cutoff}
return False
async def _poll_loop(self) -> None:
while True:
event = await self._message_queue.get()
plugins/platforms/whatsapp/adapter.py
+2 -4
View File
@@ -176,6 +176,7 @@ from gateway.whatsapp_identity import to_whatsapp_jid
from gateway.platforms.base import (
BasePlatformAdapter, SendResult, SUPPORTED_DOCUMENT_TYPES, cache_image_from_url, cache_audio_from_url,
)
from gateway.platforms.helpers import cancel_task
from gateway.platforms.event import MessageEvent, MessageType
from utils import env_int
@@ -538,10 +539,7 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
except Exception as e:
print(f"[{self.name}] Error stopping bridge: {e}")
_unlink_quietly(self._session_path / "bridge.pid")
if self._poll_task and not self._poll_task.done():
self._poll_task.cancel()
with suppress(asyncio.CancelledError, Exception):
await self._poll_task
await cancel_task(self._poll_task)
if self._http_session and not self._http_session.closed:
await self._http_session.close()
self._poll_task = self._http_session = self._bridge_process = None
tests/gateway/test_line_plugin.py
+1 -2
View File
@@ -45,7 +45,6 @@ check_requirements = _line.check_requirements
validate_config = _line.validate_config
_standalone_send = _line._standalone_send
_env_enablement = _line._env_enablement
_MessageDeduplicator = _line._MessageDeduplicator
# ---------------------------------------------------------------------------
@@ -112,7 +111,7 @@ class TestAllowlist:
class TestDedup:
def test_first_event_not_duplicate(self):
d = _MessageDeduplicator()
d = _line.MessageDeduplicator(max_size=1000, ttl_seconds=float("inf"))
assert not d.is_duplicate("evt1")
tests/gateway/test_ntfy_plugin.py
+3 -3
View File
@@ -139,12 +139,12 @@ class TestDeduplication:
def test_first_message_not_duplicate(self):
adapter = self._make_adapter()
assert adapter._is_duplicate("msg-1") is False
assert adapter._dedup.is_duplicate("msg-1") is False
def test_second_occurrence_is_duplicate(self):
adapter = self._make_adapter()
adapter._is_duplicate("msg-1")
assert adapter._is_duplicate("msg-1") is True
adapter._dedup.is_duplicate("msg-1")
assert adapter._dedup.is_duplicate("msg-1") is True
# ---------------------------------------------------------------------------
tests/gateway/test_shared_adapter_utils.py
+96
View File
@@ -0,0 +1,96 @@
"""Adapters share the three tiny lifecycle utilities in ``gateway.platforms.helpers`` instead of
re-declaring them: ``cancel_task`` (cancel + await, self-cancel safe), ``MessageDeduplicator``
(TTL + size bound) and ``bounded_put`` (insertion-ordered hard cap).
"""
from __future__ import annotations
import asyncio
import inspect
from collections import OrderedDict
import pytest
from gateway.platforms import helpers
def _adapter_classes():
from gateway.platforms.qqbot.adapter import QQAdapter
from gateway.platforms.signal import SignalAdapter
from gateway.platforms.whatsapp_cloud import WhatsAppCloudAdapter
from plugins.platforms.buzz.adapter import BuzzAdapter
from plugins.platforms.ntfy.adapter import NtfyAdapter
from plugins.platforms.photon.adapter import PhotonAdapter
from plugins.platforms.simplex.adapter import SimplexAdapter
from plugins.platforms.wecom.callback_adapter import WecomCallbackAdapter
return [QQAdapter, SignalAdapter, WhatsAppCloudAdapter, BuzzAdapter, NtfyAdapter,
PhotonAdapter, SimplexAdapter, WecomCallbackAdapter]
def test_no_adapter_redeclares_the_shared_utilities():
"""The bodies used to live as `_cancel_task` / `_is_duplicate` / `_bounded_put` on each class."""
for cls in _adapter_classes():
module = inspect.getmodule(cls)
for name in ("_cancel_task", "_is_duplicate", "_bounded_put"):
assert not hasattr(cls, name), (cls.__name__, name)
assert not hasattr(module, name), (module.__name__, name)
def test_cancel_task_unwinds_and_is_self_cancel_safe():
async def scenario():
started = asyncio.Event()
async def worker():
started.set()
await asyncio.sleep(60)
task = asyncio.create_task(worker())
await started.wait()
await helpers.cancel_task(task)
assert task.cancelled()
await helpers.cancel_task(None)
await helpers.cancel_task(task) # already done: no-op
async def suicidal():
await helpers.cancel_task(asyncio.current_task())
return "survived"
me = asyncio.create_task(suicidal())
with pytest.raises(asyncio.CancelledError):
await me
asyncio.run(scenario())
def test_bounded_put_refreshes_and_caps():
store: OrderedDict = OrderedDict()
for i in range(5):
helpers.bounded_put(store, f"k{i}", i, cap=3)
assert list(store) == ["k2", "k3", "k4"]
helpers.bounded_put(store, "k2", "again", cap=3) # refresh moves it to the newest slot
assert list(store) == ["k3", "k4", "k2"]
def test_dedup_sites_keep_their_own_window(monkeypatch):
from gateway.platforms.qqbot import constants as qq
from plugins.platforms.photon import adapter as photon
from plugins.platforms.wecom import callback_adapter as wecom_cb
import plugins.platforms.line.adapter as line
windows = {
"qqbot": (qq.DEDUP_MAX_SIZE, qq.DEDUP_WINDOW_SECONDS),
"photon": (photon._DEDUP_MAX_SIZE, photon._DEDUP_WINDOW_SECONDS),
"wecom_callback": (2000, wecom_cb.MESSAGE_DEDUP_TTL_SECONDS),
"line": (1000, float("inf")),
}
# The helper honours ttl exactly: a stale id is admitted again, a fresh one is not.
now = [1000.0]
monkeypatch.setattr(helpers.time, "time", lambda: now[0])
for name, (cap, ttl) in windows.items():
d = helpers.MessageDeduplicator(max_size=cap, ttl_seconds=ttl)
assert d.is_duplicate("m") is False and d.is_duplicate("m") is True, name
now[0] += ttl if ttl != float("inf") else 10 ** 9
assert d.is_duplicate("m") is (ttl == float("inf")), name
assert line.MessageDeduplicator is helpers.MessageDeduplicator
tests/gateway/test_whatsapp_cloud.py
+3 -5
View File
@@ -1332,14 +1332,12 @@ class TestBoundedInteractiveState:
def test_bounded_put_evicts_oldest(self):
from collections import OrderedDict
from gateway.platforms.whatsapp_cloud import (
INTERACTIVE_STATE_CACHE_SIZE,
WhatsAppCloudAdapter,
)
from gateway.platforms.helpers import bounded_put
from gateway.platforms.whatsapp_cloud import INTERACTIVE_STATE_CACHE_SIZE
cache: OrderedDict = OrderedDict()
for i in range(INTERACTIVE_STATE_CACHE_SIZE + 10):
WhatsAppCloudAdapter._bounded_put(cache, f"id-{i}", "sess")
bounded_put(cache, f"id-{i}", "sess", INTERACTIVE_STATE_CACHE_SIZE)
assert len(cache) == INTERACTIVE_STATE_CACHE_SIZE
assert "id-0" not in cache
assert f"id-{INTERACTIVE_STATE_CACHE_SIZE + 9}" in cache
tests/plugins/platforms/photon/test_inbound.py
+5 -5
View File
@@ -1,7 +1,7 @@
"""Inbound dispatch + dedup tests for PhotonAdapter.
These bypass the loopback HTTP stream — they call ``_dispatch_inbound`` /
``_on_inbound_line`` / ``_is_duplicate`` directly, exercising the
``_on_inbound_line`` / ``_dedup`` directly, exercising the
sidecar-event parsing without spawning the Node sidecar or binding ports.
"""
from __future__ import annotations
@@ -152,10 +152,10 @@ async def test_on_inbound_line_dispatches_and_dedups(
def test_is_duplicate_window(monkeypatch: pytest.MonkeyPatch) -> None:
adapter = _make_adapter(monkeypatch)
assert adapter._is_duplicate("id-1") is False
assert adapter._is_duplicate("id-1") is True
assert adapter._is_duplicate("id-2") is False
assert adapter._is_duplicate("id-1") is True # still dup
assert adapter._dedup.is_duplicate("id-1") is False
assert adapter._dedup.is_duplicate("id-1") is True
assert adapter._dedup.is_duplicate("id-2") is False
assert adapter._dedup.is_duplicate("id-1") is True # still dup
def test_check_requirements_without_node(monkeypatch: pytest.MonkeyPatch) -> None: