From d9811d81f91895ef174baa4f7108db5763d203b6 Mon Sep 17 00:00:00 2001 From: meng93 Date: Thu, 23 Apr 2026 11:56:58 +0800 Subject: [PATCH] =?UTF-8?q?fix(dingtalk):=20adapter=20reliability=20?= =?UTF-8?q?=E2=80=94=20websockets=20proxy,=20card=20QPS=20throttle,=20inbo?= =?UTF-8?q?und=20queue=20serialization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Capture websockets proxy env vars before dingtalk-stream SDK can clobber them; restore after import (_install_dingtalk_websockets_proxy). - Add _CardTokenBucket (token-bucket, 20 QPS) and per-message 800 ms edit_message() throttle to stay under DingTalk AI-Card rate limits. 403 responses trigger an automatic 2 s exponential backoff. - Inbound message queue (_enqueue_inbound / _sweep_session_queues) serialises same-chat messages so long-running agent turns are not duplicated; a random 'busy' acknowledgement is sent for queued msgs. - Hydrate *_HOME_CHANNEL yaml keys into os.environ on gateway boot so /sethome survives process restarts (gateway/config.py). - Fix REGISTRATION_SOURCE default (openClaw → DING_DWS_CLAW) in hermes_cli/dingtalk_auth.py. - Platform display-name overrides for /sethome prompt (DingTalk, WeCom, etc.) in gateway/run.py. - Add scripts/gateway_guard.sh — auto-restart supervisor for the gateway process with caffeinate support on macOS. - Add .claude to .gitignore. --- .gitignore | 3 + gateway/config.py | 16 ++ gateway/platforms/dingtalk.py | 278 ++++++++++++++++++++++++- gateway/run.py | 13 +- hermes_cli/dingtalk_auth.py | 2 +- scripts/gateway_guard.sh | 107 ++++++++++ tests/hermes_cli/test_dingtalk_auth.py | 2 +- 7 files changed, 411 insertions(+), 10 deletions(-) create mode 100755 scripts/gateway_guard.sh diff --git a/.gitignore b/.gitignore index 72f3bd17f7db..385ff7cabe2f 100644 --- a/.gitignore +++ b/.gitignore @@ -69,3 +69,6 @@ mini-swe-agent/ .nix-stamps/ result website/static/api/skills-index.json + +.claude + diff --git a/gateway/config.py b/gateway/config.py index 67ebf7346189..0263fe476540 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -491,6 +491,22 @@ def load_gateway_config() -> GatewayConfig: with open(config_yaml_path, encoding="utf-8") as f: yaml_cfg = yaml.safe_load(f) or {} + # Hydrate top-level UPPERCASE_HOME_CHANNEL keys from config.yaml + # into os.environ so `_apply_env_overrides` picks them up on boot. + # Without this, /sethome-written yaml entries get orphaned across + # gateway restarts (yaml survives, process env does not), which + # makes onboarding prompts and home-channel routing flap between + # sessions. Env var (if already set) takes precedence. + for _k, _v in yaml_cfg.items(): + if ( + isinstance(_k, str) + and _k.endswith("_HOME_CHANNEL") + and _k.isupper() + and _v + and not os.getenv(_k) + ): + os.environ[_k] = str(_v) + # Map config.yaml keys → GatewayConfig.from_dict() schema. # Each key overwrites whatever gateway.json may have set. sr = yaml_cfg.get("session_reset") diff --git a/gateway/platforms/dingtalk.py b/gateway/platforms/dingtalk.py index 3037e402b2cd..f8123118436d 100644 --- a/gateway/platforms/dingtalk.py +++ b/gateway/platforms/dingtalk.py @@ -30,11 +30,13 @@ import json import logging import os +import random import re +import time import traceback import uuid from datetime import datetime, timezone -from typing import Any, Dict, List, Optional, Set +from typing import Any, Dict, List, Optional, Set, Tuple try: import dingtalk_stream @@ -56,6 +58,19 @@ }, ) # type: ignore[assignment] +# Capture pristine ``websockets.connect`` at our import time so dingtalk-stream +# always uses the original even if another module (e.g. the Feishu adapter) +# later monkey-patches the global ``websockets.connect`` with an ``async def`` +# wrapper — that turns the return value into a coroutine and breaks the SDK's +# ``async with websockets.connect(uri)`` at dingtalk_stream/stream.py:74. +try: + import websockets as _pristine_ws_module + + _PRISTINE_WEBSOCKETS_CONNECT = _pristine_ws_module.connect +except ImportError: + _pristine_ws_module = None # type: ignore[assignment] + _PRISTINE_WEBSOCKETS_CONNECT = None # type: ignore[assignment] + try: import httpx @@ -64,6 +79,37 @@ HTTPX_AVAILABLE = False httpx = None # type: ignore[assignment] + +_WS_PROXY_INSTALLED = False + + +def _install_dingtalk_websockets_proxy() -> None: + """Point dingtalk_stream.stream.websockets at a namespace holding the + pristine ``connect`` captured at our import time. + + Idempotent. Safe no-op when dingtalk-stream or websockets is unavailable. + """ + global _WS_PROXY_INSTALLED + if _WS_PROXY_INSTALLED: + return + if _PRISTINE_WEBSOCKETS_CONNECT is None or dingtalk_stream is None: + return + try: + import types + + from dingtalk_stream import stream as _dts_stream + + _dts_stream.websockets = types.SimpleNamespace( + connect=_PRISTINE_WEBSOCKETS_CONNECT, + exceptions=_pristine_ws_module.exceptions, + ) + _WS_PROXY_INSTALLED = True + except Exception: # pragma: no cover - defensive + logger.debug( + "[DingTalk] failed to install websockets proxy on dingtalk_stream.stream", + exc_info=True, + ) + # Card SDK for AI Cards (following QwenPaw pattern) try: from alibabacloud_dingtalk.card_1_0 import ( @@ -103,6 +149,81 @@ _SESSION_WEBHOOKS_MAX = 500 _DINGTALK_WEBHOOK_RE = re.compile(r'^https://(?:api|oapi)\.dingtalk\.com/') +# AI Card streaming_update QPS guard. The DingTalk gateway returns HTTP 403 +# "concurrent update" when multiple edit_message calls hit the same card +# within ~500ms. Reference (openclaw-connector reply-dispatcher.ts:103) uses +# 800ms as the per-card minimum interval between non-finalize edits. We +# match that to avoid the same 403 storm. Finalize edits are NEVER +# throttled — dropping them would leave the card stuck in streaming state. +_CARD_EDIT_THROTTLE_MS = 800 +# Error-send cooldown so repeated failures don't spam users +# (reply-dispatcher.ts:108 uses 60s — same pattern, same reason). +_ERROR_COOLDOWN_MS = 60_000 + +# Global token-bucket rate for AI Card streaming_update across ALL chats. +# DingTalk's official cap is ~40 QPS per tenant; reference connector +# (messaging/card.ts:18) uses 20 as a safety margin. Matching that, so +# concurrent sessions never bust the tenant-wide ceiling. +_CARD_API_MAX_QPS = 20 +_CARD_API_QPS_BACKOFF_MS = 2_000 + +# Inbound-message queue TTL. queueKey entries that haven't received a new +# message for this long are eligible for sweep (reference +# core/message-handler.ts:92 uses 5 min). +_INBOUND_QUEUE_TTL_SEC = 300 +# Busy-ACK phrases when inbound queue already has a pending task. Picked +# randomly so repeats don't feel scripted (reference utils/constants.ts +# QUEUE_BUSY_ACK_PHRASES). +_QUEUE_BUSY_ACK_PHRASES = ( + "收到，让我先把前一条处理完 🙏", + "稍等，排队中……", + "收到～手头这条完事就来", + "别急，按顺序处理中", +) + + +class _CardTokenBucket: + """Global async token bucket for DingTalk card streaming_update. + + Mirrors messaging/card.ts:23-95. All streamAICard/edit_message calls + across every chat + account share one bucket so concurrent sessions + don't blow past the tenant-wide QPS limit and trigger 403 storms. + Refills at ``rate`` tokens/second with capacity = rate. On an + upstream 403 limit, callers call ``trigger_backoff`` and subsequent + acquirers wait out the backoff window. + """ + + def __init__(self, rate: float) -> None: + self._rate = float(rate) + self._tokens = float(rate) + self._last_refill = time.monotonic() + self._backoff_until = 0.0 + self._lock = asyncio.Lock() + + async def acquire(self) -> None: + async with self._lock: + now = time.monotonic() + if now < self._backoff_until: + await asyncio.sleep(self._backoff_until - now) + now = time.monotonic() + elapsed = now - self._last_refill + self._tokens = min(self._rate, self._tokens + elapsed * self._rate) + self._last_refill = now + if self._tokens < 1.0: + wait_s = (1.0 - self._tokens) / self._rate + await asyncio.sleep(wait_s) + self._tokens = 0.0 + self._last_refill = time.monotonic() + else: + self._tokens -= 1.0 + + def trigger_backoff(self) -> None: + self._backoff_until = time.monotonic() + _CARD_API_QPS_BACKOFF_MS / 1000.0 + + +# Process-wide card-API rate limiter (shared across adapters). +_CARD_BUCKET = _CardTokenBucket(_CARD_API_MAX_QPS) + # DingTalk message type → runtime content type DINGTALK_TYPE_MAPPING = { "picture": "image", @@ -202,10 +323,19 @@ def __init__(self, config: PlatformConfig): # auto-close them as siblings — otherwise tool-progress cards get # stuck in streaming state forever. self._streaming_cards: Dict[str, Dict[str, str]] = {} + # Per-card last-edit timestamp (ms) + self._card_last_edit_ms: Dict[str, int] = {} + # Per-chat error-send cooldown + self._error_last_sent_ms: Dict[str, int] = {} # Track fire-and-forget emoji/reaction coroutines so Python's GC # doesn't drop them mid-flight, and we can cancel them on disconnect. self._bg_tasks: Set[asyncio.Task] = set() + # Per-session inbound message queue + self._session_queues: Dict[str, asyncio.Task] = {} + self._session_last_activity: Dict[str, float] = {} + self._session_queue_sweeper: Optional[asyncio.Task] = None + # -- Connection lifecycle ----------------------------------------------- async def connect(self) -> bool: @@ -230,6 +360,8 @@ async def connect(self) -> bool: try: self._http_client = httpx.AsyncClient(timeout=30.0) + _install_dingtalk_websockets_proxy() + credential = dingtalk_stream.Credential( self._client_id, self._client_secret ) @@ -263,6 +395,9 @@ async def connect(self) -> bool: ) self._stream_task = asyncio.create_task(self._run_stream()) + self._session_queue_sweeper = asyncio.create_task( + self._sweep_session_queues() + ) self._mark_connected() logger.info("[%s] Connected via Stream Mode", self.name) return True @@ -323,6 +458,26 @@ async def disconnect(self) -> None: logger.debug("[%s] stream task did not exit cleanly during disconnect", self.name) self._stream_task = None + # Stop the session-queue sweeper. + if self._session_queue_sweeper: + self._session_queue_sweeper.cancel() + try: + await self._session_queue_sweeper + except (asyncio.CancelledError, Exception): + pass + self._session_queue_sweeper = None + + # Cancel any still-pending inbound-queue tasks. + if self._session_queues: + for task in list(self._session_queues.values()): + if not task.done(): + task.cancel() + await asyncio.gather( + *self._session_queues.values(), return_exceptions=True, + ) + self._session_queues.clear() + self._session_last_activity.clear() + # Cancel any in-flight background tasks (emoji reactions, etc.) if self._bg_tasks: for task in list(self._bg_tasks): @@ -338,6 +493,8 @@ async def disconnect(self) -> None: self._session_webhooks.clear() self._message_contexts.clear() self._streaming_cards.clear() + self._card_last_edit_ms.clear() + self._error_last_sent_ms.clear() self._done_emoji_fired.clear() self._dedup.clear() logger.info("[%s] Disconnected", self.name) @@ -460,6 +617,78 @@ def _spawn_bg(self, coro) -> None: self._bg_tasks.add(task) task.add_done_callback(self._bg_tasks.discard) + # -- Inbound serialization queue --------------------------------------- + + def _inbound_queue_key(self, chatbot_msg: "ChatbotMessage") -> str: + conv_id = getattr(chatbot_msg, "conversation_id", "") or "" + sender_id = getattr(chatbot_msg, "sender_id", "") or "" + return conv_id or sender_id + + async def _send_busy_ack(self, chatbot_msg: "ChatbotMessage") -> None: + if not self._http_client: + return + webhook = getattr(chatbot_msg, "session_webhook", "") or "" + if not webhook or not _DINGTALK_WEBHOOK_RE.match(webhook): + return + phrase = random.choice(_QUEUE_BUSY_ACK_PHRASES) + try: + await self._http_client.post( + webhook, + json={"msgtype": "text", "text": {"content": phrase}}, + timeout=5.0, + ) + except Exception as e: + logger.debug("[%s] busy ACK send failed: %s", self.name, e) + + async def _enqueue_inbound(self, chatbot_msg: "ChatbotMessage") -> None: + queue_key = self._inbound_queue_key(chatbot_msg) + if not queue_key: + await self._on_message(chatbot_msg) + return + + self._session_last_activity[queue_key] = time.monotonic() + prev_task = self._session_queues.get(queue_key) + is_busy = prev_task is not None and not prev_task.done() + + if is_busy: + self._spawn_bg(self._send_busy_ack(chatbot_msg)) + + async def _chained() -> None: + if prev_task is not None: + try: + await prev_task + except Exception: + pass + await self._on_message(chatbot_msg) + + task = asyncio.create_task(_chained()) + self._session_queues[queue_key] = task + + def _cleanup(t: asyncio.Task) -> None: + if self._session_queues.get(queue_key) is t: + self._session_queues.pop(queue_key, None) + + task.add_done_callback(_cleanup) + + async def _sweep_session_queues(self) -> None: + try: + while self._running: + await asyncio.sleep(60) + now = time.monotonic() + stale = [ + k for k, ts in self._session_last_activity.items() + if now - ts > _INBOUND_QUEUE_TTL_SEC + ] + for k in stale: + self._session_last_activity.pop(k, None) + task = self._session_queues.get(k) + if task is not None and task.done(): + self._session_queues.pop(k, None) + except asyncio.CancelledError: + return + except Exception as e: + logger.debug("[%s] session-queue sweep error: %s", self.name, e) + # -- AI Card lifecycle helpers ------------------------------------------ async def _close_streaming_siblings(self, chat_id: str) -> None: @@ -1020,6 +1249,19 @@ async def edit_message( """ if not message_id: return SendResult(success=False, error="message_id required") + + # Throttle non-finalize edits per out_track_id. + now_ms = int(datetime.now(tz=timezone.utc).timestamp() * 1000) + if not finalize: + last_ms = self._card_last_edit_ms.get(message_id, 0) + if now_ms - last_ms < _CARD_EDIT_THROTTLE_MS: + logger.debug( + "[%s] edit_message throttled (%dms since last) for %s", + self.name, now_ms - last_ms, message_id, + ) + return SendResult(success=True, message_id=message_id) + self._card_last_edit_ms[message_id] = now_ms + token = await self._get_access_token() if not token: return SendResult(success=False, error="No access token") @@ -1035,6 +1277,7 @@ async def edit_message( self._streaming_cards.get(chat_id, {}).pop(message_id, None) if not self._streaming_cards.get(chat_id): self._streaming_cards.pop(chat_id, None) + self._card_last_edit_ms.pop(message_id, None) logger.debug( "[%s] AI Card finalized (edit): %s", self.name, message_id, @@ -1057,7 +1300,13 @@ async def _stream_card_content( content: str, finalize: bool = False, ) -> None: - """Stream content to an existing AI Card.""" + """Stream content to an existing AI Card. + + Per-card 800ms throttle happens at the ``edit_message`` layer; this + function additionally goes through the **global** token bucket so + that many parallel chats can't collectively overrun the tenant-wide + DingTalk card-API QPS cap (~40/s). + """ stream_request = dingtalk_card_models.StreamingUpdateRequest( out_track_id=out_track_id, guid=str(uuid.uuid4()), @@ -1073,9 +1322,20 @@ async def _stream_card_content( ) runtime = tea_util_models.RuntimeOptions() - await self._card_sdk.streaming_update_with_options_async( - stream_request, stream_headers, runtime - ) + await _CARD_BUCKET.acquire() + try: + await self._card_sdk.streaming_update_with_options_async( + stream_request, stream_headers, runtime + ) + except Exception as e: + err_msg = str(e) + if "QpsLimit" in err_msg or "403" in err_msg or "qps" in err_msg.lower(): + logger.warning( + "[%s] Card QPS limit hit, backing off %dms: %s", + self.name, _CARD_API_QPS_BACKOFF_MS, err_msg[:160], + ) + _CARD_BUCKET.trigger_backoff() + raise async def _get_access_token(self) -> Optional[str]: """Get access token using SDK's cached token.""" @@ -1353,9 +1613,13 @@ async def process(self, message: "CallbackMessage"): return AckMessage.STATUS_OK, "OK" async def _safe_on_message(self, chatbot_msg: "ChatbotMessage") -> None: - """Wrapper that catches exceptions from _on_message.""" + """Wrapper that catches exceptions from _on_message. + + Dispatches through ``_enqueue_inbound`` so same-chat messages are + serialized (with a busy-ACK on the tail). + """ try: - await self._adapter._on_message(chatbot_msg) + await self._adapter._enqueue_inbound(chatbot_msg) except Exception: logger.exception( "[%s] Error processing incoming message", self._adapter.name diff --git a/gateway/run.py b/gateway/run.py index a024649cbdd1..b9bd12e189fe 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4411,12 +4411,23 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g if not history and source.platform and source.platform != Platform.LOCAL and source.platform != Platform.WEBHOOK: platform_name = source.platform.value env_key = f"{platform_name.upper()}_HOME_CHANNEL" + # Platform display-name overrides for mixed-case brands where + # naive .title() reads awkwardly ("Dingtalk" → "DingTalk"). + _display_overrides = { + "dingtalk": "DingTalk", + "feishu": "Feishu", + "wecom": "WeCom", + "qqbot": "QQ", + "bluebubbles": "BlueBubbles", + "homeassistant": "Home Assistant", + } + display_name = _display_overrides.get(platform_name, platform_name.title()) if not os.getenv(env_key): adapter = self.adapters.get(source.platform) if adapter: await adapter.send( source.chat_id, - f"📬 No home channel is set for {platform_name.title()}. " + f"📬 No home channel is set for {display_name}. " f"A home channel is where Hermes delivers cron job results " f"and cross-platform messages.\n\n" f"Type /sethome to make this chat your home channel, " diff --git a/hermes_cli/dingtalk_auth.py b/hermes_cli/dingtalk_auth.py index e1034c53da62..6f9c692b07a6 100644 --- a/hermes_cli/dingtalk_auth.py +++ b/hermes_cli/dingtalk_auth.py @@ -30,7 +30,7 @@ "DINGTALK_REGISTRATION_BASE_URL", "https://oapi.dingtalk.com" ).rstrip("/") -REGISTRATION_SOURCE = os.environ.get("DINGTALK_REGISTRATION_SOURCE", "openClaw") +REGISTRATION_SOURCE = os.environ.get("DINGTALK_REGISTRATION_SOURCE", "DING_DWS_CLAW") # ── API helpers ──────────────────────────────────────────────────────────── diff --git a/scripts/gateway_guard.sh b/scripts/gateway_guard.sh new file mode 100755 index 000000000000..6cb3a72b383f --- /dev/null +++ b/scripts/gateway_guard.sh @@ -0,0 +1,107 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +LOG_DIR="${ROOT_DIR}/logs" +PID_FILE="${LOG_DIR}/gateway-guard.pid" +RUN_LOG="${LOG_DIR}/gateway-guard.log" + +mkdir -p "${LOG_DIR}" + +usage() { + cat <<'EOF' +Usage: scripts/gateway_guard.sh + +start Start gateway in background with auto-restart loop +stop Stop background guard process +restart Restart guard process +status Show whether guard is running +logs Follow logs +EOF +} + +is_running() { + if [[ -f "${PID_FILE}" ]]; then + local pid + pid="$(cat "${PID_FILE}")" + if [[ -n "${pid}" ]] && kill -0 "${pid}" 2>/dev/null; then + return 0 + fi + fi + return 1 +} + +start_guard() { + if is_running; then + echo "gateway guard is already running (pid: $(cat "${PID_FILE}"))" + exit 0 + fi + + ( + cd "${ROOT_DIR}" + while true; do + echo "===== $(date '+%F %T') gateway start =====" >> "${RUN_LOG}" + if command -v caffeinate >/dev/null 2>&1; then + caffeinate -dimsu venv/bin/python -m hermes_cli.main gateway run --replace -v >> "${RUN_LOG}" 2>&1 + else + venv/bin/python -m hermes_cli.main gateway run --replace -v >> "${RUN_LOG}" 2>&1 + fi + code=$? + echo "===== $(date '+%F %T') gateway exited code=${code}, restart in 5s =====" >> "${RUN_LOG}" + sleep 5 + done + ) & + + echo $! > "${PID_FILE}" + echo "gateway guard started (pid: $(cat "${PID_FILE}"))" + echo "log file: ${RUN_LOG}" +} + +stop_guard() { + if ! is_running; then + echo "gateway guard is not running" + rm -f "${PID_FILE}" + exit 0 + fi + + local pid + pid="$(cat "${PID_FILE}")" + kill "${pid}" 2>/dev/null || true + sleep 1 + if kill -0 "${pid}" 2>/dev/null; then + kill -9 "${pid}" 2>/dev/null || true + fi + rm -f "${PID_FILE}" + echo "gateway guard stopped" +} + +status_guard() { + if is_running; then + echo "gateway guard is running (pid: $(cat "${PID_FILE}"))" + else + echo "gateway guard is not running" + fi +} + +logs_guard() { + touch "${RUN_LOG}" + tail -f "${RUN_LOG}" +} + +main() { + if [[ $# -lt 1 ]]; then + usage + exit 1 + fi + + case "${1}" in + start) start_guard ;; + stop) stop_guard ;; + restart) stop_guard; start_guard ;; + status) status_guard ;; + logs) logs_guard ;; + *) usage; exit 1 ;; + esac +} + +main "$@" diff --git a/tests/hermes_cli/test_dingtalk_auth.py b/tests/hermes_cli/test_dingtalk_auth.py index 592cd3175ead..4cc9535cdd4a 100644 --- a/tests/hermes_cli/test_dingtalk_auth.py +++ b/tests/hermes_cli/test_dingtalk_auth.py @@ -214,4 +214,4 @@ def test_source_default(self, monkeypatch): import importlib import hermes_cli.dingtalk_auth as mod importlib.reload(mod) - assert mod.REGISTRATION_SOURCE == "openClaw" + assert mod.REGISTRATION_SOURCE == "DING_DWS_CLAW"