diff --git a/tools/react_to_message_tool.py b/tools/react_to_message_tool.py index fa37347fda..4859d25f38 100644 --- a/tools/react_to_message_tool.py +++ b/tools/react_to_message_tool.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 """Let the agent react to a message with an emoji in the Hermes desktop app. -The conversational counterpart to the user's tapback: same reaction store, -same one-per-author semantics, written with ``author="agent"``. Lives in the -``desktop_ui`` toolset so it costs nothing on other surfaces (platform adapters -expose reactions via ``send_message(action="react")``). Defaults to the message -that triggered this turn so the model needn't thread row ids through tool calls, -and emits ``message.reaction`` so the renderer paints it without a resume. +The conversational counterpart to the user's tapback: same reaction store, same +one-per-author semantics, written with ``author="agent"``. Lives in the ``desktop_ui`` +toolset so it costs nothing on other surfaces (platform adapters expose reactions via +``send_message(action="react")``). Defaults to the message that triggered this turn so +the model needn't thread row ids through tool calls, and emits ``message.reaction`` so +the renderer paints it without a resume. """ import json @@ -30,15 +30,14 @@ def _react(emoji: str, message_row_id, messages_back, *, db, session_key: str) - row_id = message_row_id target_role = "user" if row_id is None: - # Default target: the latest user message. `messages_back` steps to earlier - # user turns (1 = the one before) — ids aren't visible to the model and - # quoting text is ambiguous, but "two messages ago" is how a person thinks. + # Default target: the latest user message. `messages_back` steps to earlier user + # turns (1 = the one before) — ids aren't visible to the model and quoting text is + # ambiguous, but "two messages ago" is how a person thinks. back = max(0, int(messages_back or 0)) row_id = db.latest_message_row_id(session_key, role="user", offset=back) if row_id is None: return tool_error( - f"No user message found {back} back." if back else "No user message to react to yet." - ) + f"No user message found {back} back." if back else "No user message to react to yet.") else: target_role = db.get_message_role(session_key, int(row_id)) or "user" @@ -49,28 +48,22 @@ def _react(emoji: str, message_row_id, messages_back, *, db, session_key: str) - if reactions is None: return tool_error(f"Message {row_id} is not part of this conversation.") - # Paint it live. A missing bridge (non-desktop surface) is not an error — the - # reaction is persisted and shows on next load. `role` lets the renderer match - # a live message that doesn't know its durable row id yet (learned on resume). + # Paint it live. A missing bridge (non-desktop surface) is not an error — the reaction + # is persisted and shows on next load. `role` lets the renderer match a live message + # that doesn't know its durable row id yet (learned on resume). try: desktop_ui.emit( - "message.reaction", - {"row_id": int(row_id), "reactions": reactions, "role": target_role}, - ) + "message.reaction", {"row_id": int(row_id), "reactions": reactions, "role": target_role}) except Exception: pass - return json.dumps( - {"success": True, "row_id": int(row_id), "reactions": reactions}, ensure_ascii=False - ) + return json.dumps({"success": True, "row_id": int(row_id), "reactions": reactions}, ensure_ascii=False) def react_to_message_tool(emoji: str, message_row_id=None, messages_back=None) -> str: """Attach (or with an empty ``emoji`` retract) the agent's reaction.""" emoji = (emoji or "").strip() - session_key = get_session_env("HERMES_SESSION_KEY", "") or get_session_env( - "HERMES_SESSION_ID", "" - ) + session_key = get_session_env("HERMES_SESSION_KEY", "") or get_session_env("HERMES_SESSION_ID", "") if not session_key: return tool_error("No active session — reactions need a persisted conversation.") diff --git a/tools/send_message_senders.py b/tools/send_message_senders.py index 556c859a0e..68bbe156f3 100644 --- a/tools/send_message_senders.py +++ b/tools/send_message_senders.py @@ -14,67 +14,52 @@ _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".gif"} _VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"} _AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".m2a", ".wav", ".m4a", ".flac"} _VOICE_EXTS = {".ogg", ".opus"} -# Telegram's Bot API sendAudio only accepts MP3 / M4A. Other audio -# formats either route through sendVoice (Opus/OGG) or fall back to -# document delivery. +# Telegram's sendAudio only accepts MP3 / M4A; other audio goes via sendVoice +# (Opus/OGG) or falls back to document delivery. _TELEGRAM_SEND_AUDIO_EXTS = {".mp3", ".m4a"} -# Extensions that carry a native caption on the media bubble itself -# (photo/video/document). Voice/audio notes are excluded: a caption on a -# voice note reads as a separate label rather than a bubble caption, and the -# established convention is to keep the accompanying text as its own message. -_CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | { - ".pdf", ".doc", ".docx", ".txt", ".md", ".csv", ".xlsx", ".zip", -} +# Extensions that carry a native caption on the media bubble (photo/video/document). +# Voice/audio notes are excluded: a caption on a voice note reads as a separate label, +# so the accompanying text stays its own message. +_CAPTIONABLE_EXTS = _IMAGE_EXTS | _VIDEO_EXTS | {".pdf", ".doc", ".docx", ".txt", ".md", ".csv", ".xlsx", ".zip"} -# Per-platform native caption length limits (characters). Text longer than -# the limit can't ride on the media bubble and stays a separate body message. -# Telegram's photo/video caption cap is 1024; WhatsApp and Discord are far -# more generous, so a conservative shared ceiling keeps behavior predictable. +# Native caption length limits (characters). Telegram's photo/video cap is 1024; +# WhatsApp/Discord are far more generous, so one conservative shared ceiling. _TELEGRAM_CAPTION_LIMIT = 1024 _DEFAULT_CAPTION_LIMIT = 4096 def _media_caption_split(text, media_files, *, max_caption_len): - """Decide whether the accompanying text rides on the media bubble. + """Single chokepoint deciding whether text rides on the media bubble as its caption. - Single chokepoint for ``MEDIA: caption`` across every sender. - Returns ``(caption, "")`` — text becomes the native caption, no separate - body message — only for exactly one media file of a captionable kind - (image/video/document, not a voice/audio note) whose text fits - ``max_caption_len``. Otherwise ``(None, text)``: text goes as a separate - message (multi-file caption→file association is ambiguous). - - Length is measured in codepoints: never under-counts Telegram's UTF-16 - units for BMP text, so over-counting only fails safe. The Telegram sender - re-checks the *formatted* caption since escaping inflates it. + ``(caption, "")`` only for exactly one captionable file (not a voice/audio note) + whose text fits ``max_caption_len``; otherwise ``(None, text)`` — multi-file + caption→file association is ambiguous. Length is codepoints, which never + under-counts Telegram's UTF-16 units for BMP text (over-counting fails safe); + the Telegram sender re-checks the *formatted* caption since escaping inflates it. """ stripped = (text or "").strip() media = media_files or [] - if not stripped or len(media) != 1: + if not stripped or len(media) != 1 or len(stripped) > max_caption_len: return None, text media_path, is_voice = media[0] if is_voice or os.path.splitext(media_path)[1].lower() not in _CAPTIONABLE_EXTS: return None, text - if len(stripped) > max_caption_len: - return None, text return stripped, "" + + _URL_SECRET_QUERY_RE = re.compile( r"([?&](?:access_token|api[_-]?key|auth[_-]?token|token|signature|sig)=)([^&#\s]+)", - re.IGNORECASE, -) + re.IGNORECASE) _GENERIC_SECRET_ASSIGN_RE = re.compile( - r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)", - re.IGNORECASE, -) + r"\b(access_token|api[_-]?key|auth[_-]?token|signature|sig)\s*=\s*([^\s,;]+)", re.IGNORECASE) def _sanitize_error_text(text) -> str: """Redact secrets from error text before surfacing it to users/models.""" redacted = redact_sensitive_text(text) redacted = _URL_SECRET_QUERY_RE.sub(lambda m: f"{m.group(1)}***", redacted) - redacted = _GENERIC_SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", redacted) - return redacted + return _GENERIC_SECRET_ASSIGN_RE.sub(lambda m: f"{m.group(1)}=***", redacted) def _error(message: str) -> dict: @@ -82,6 +67,14 @@ def _error(message: str) -> dict: return {"error": _sanitize_error_text(message)} +def _success(platform: str, chat_id, warnings=None, **fields) -> dict: + """Standard success payload; ``warnings`` is only included when non-empty.""" + result = {"success": True, "platform": platform, "chat_id": chat_id, **fields} + if warnings: + result["warnings"] = warnings + return result + + def _display_chat_id(platform_name: str, chat_id: str) -> str: """Return a result-safe chat identifier for tool transcripts/log consumers.""" if platform_name == "signal" and str(chat_id).startswith("group:"): @@ -89,16 +82,16 @@ def _display_chat_id(platform_name: str, chat_id: str) -> str: return chat_id +_NO_DELIVERABLE = "No deliverable text or media remained after processing MEDIA tags" + _TELEGRAM_TRANSIENT_MARKERS = ( "bad gateway", "502", "too many requests", "429", - "service unavailable", "503", "gateway timeout", "504", -) + "service unavailable", "503", "gateway timeout", "504") def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None: - """Seconds to wait before retrying, or None when the error is final. - Honours Telegram's ``retry_after``; timeouts are never retried (the send - may have gone through); 5xx/429 back off exponentially.""" + """Seconds to wait before retrying, or None when final. Honours ``retry_after``; + timeouts are never retried (the send may have gone through); 5xx/429 back off.""" retry_after = getattr(exc, "retry_after", None) if retry_after is not None: try: @@ -125,11 +118,7 @@ async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs) raise logger.warning( "Transient Telegram send failure (attempt %d/%d), retrying in %.1fs: %s", - attempt + 1, - attempts, - delay, - _sanitize_error_text(exc), - ) + attempt + 1, attempts, delay, _sanitize_error_text(exc)) await asyncio.sleep(delay) @@ -139,9 +128,8 @@ def _is_telegram_thread_not_found(error: Exception) -> bool: def _telegram_bot(token): - """Build a Bot honouring TELEGRAM_PROXY (config ``telegram.proxy_url``); - without it the standalone path bypasses the proxy and times out where - api.telegram.org is blocked. Falls back to a direct connection.""" + """Bot honouring TELEGRAM_PROXY (``telegram.proxy_url``) — without it the standalone + path times out where api.telegram.org is blocked. Falls back to a direct connection.""" from telegram import Bot try: @@ -153,20 +141,16 @@ def _telegram_bot(token): try: from telegram.request import HTTPXRequest logger.info("send_message: standalone Telegram send routed through proxy %s", proxy) - return Bot( - token=token, - request=HTTPXRequest(proxy=proxy), - get_updates_request=HTTPXRequest(proxy=proxy), - ) + return Bot(token=token, request=HTTPXRequest(proxy=proxy), + get_updates_request=HTTPXRequest(proxy=proxy)) except Exception as proxy_err: logger.warning("send_message: failed to attach Telegram proxy (%s), falling back to direct connection", proxy_err) return Bot(token=token) def _telegram_thread_kwargs(thread_id): - """Map a topic id to ``message_thread_id`` kwargs. Forum "General" is - thread "1" on incoming updates but Bot API rejects message_thread_id=1 - ("Message thread not found"), so it maps to no thread — same as the adapter.""" + """Topic id -> ``message_thread_id`` kwargs. Forum "General" is thread "1" inbound but + the Bot API rejects message_thread_id=1, so it maps to no thread — same as the adapter.""" if thread_id is None: return {} try: @@ -188,8 +172,8 @@ def _strip_mdv2_safe(text): async def _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, **kwargs): - """Pick the Bot API media method by extension: photo (unless forced to - document), video, voice note, sendAudio (MP3/M4A only), else document.""" + """Bot API media method by extension: photo (unless forced document), video, voice + note, sendAudio (MP3/M4A only), else document.""" if ext in _IMAGE_EXTS and not force_document: return await bot.send_photo(chat_id=chat_id, photo=f, **kwargs) if ext in _VIDEO_EXTS: @@ -202,43 +186,37 @@ async def _telegram_send_media(bot, chat_id, f, ext, is_voice, force_document, * async def _telegram_send_text_chunk(bot, chat_id, chunk, parse_mode, has_html, text_kwargs): - """Send one formatted text chunk with the adapter-matching fallbacks: - thread-not-found -> retry without ``message_thread_id`` (dropped from - ``text_kwargs`` for later chunks too); parse failure -> plain text.""" - try: + """One formatted text chunk with adapter-matching fallbacks: thread-not-found -> retry + without ``message_thread_id`` (dropped from ``text_kwargs`` for later chunks too); + parse failure -> plain text.""" + async def send(text, mode): return await _send_telegram_message_with_retry( - bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs - ) + bot, chat_id=chat_id, text=text, parse_mode=mode, **text_kwargs) + + try: + return await send(chunk, parse_mode) except Exception as md_error: if _is_telegram_thread_not_found(md_error) and text_kwargs.get("message_thread_id") is not None: logger.warning( "Thread %s not found in _send_telegram, retrying without message_thread_id", - text_kwargs.get("message_thread_id"), - ) + text_kwargs.get("message_thread_id")) text_kwargs.pop("message_thread_id", None) - return await _send_telegram_message_with_retry( - bot, chat_id=chat_id, text=chunk, parse_mode=parse_mode, **text_kwargs - ) + return await send(chunk, parse_mode) err_text = str(md_error).lower() if "parse" in err_text or "markdown" in err_text or "html" in err_text: logger.warning( "Parse mode %s failed in _send_telegram, falling back to plain text: %s", - parse_mode, - _sanitize_error_text(md_error), - ) - plain = chunk if has_html else _strip_mdv2_safe(chunk) - return await _send_telegram_message_with_retry( - bot, chat_id=chat_id, text=plain, parse_mode=None, **text_kwargs - ) + parse_mode, _sanitize_error_text(md_error)) + return await send(chunk if has_html else _strip_mdv2_safe(chunk), None) raise async def _telegram_send_one_media( bot, chat_id, media_path, is_voice, *, caption, parse_mode, has_html, thread_kwargs, force_document ): - """Upload one file with the adapter-matching fallbacks (thread-not-found -> - no ``message_thread_id``; caption parse failure -> plain caption). Each - retry re-seeks the file because the first attempt consumed it.""" + """Upload one file with adapter-matching fallbacks (thread-not-found -> no + ``message_thread_id``; caption parse failure -> plain caption). Retries re-seek + the file because the first attempt consumed it.""" ext = os.path.splitext(media_path)[1].lower() voice_note = ext in _VOICE_EXTS and is_voice with open(media_path, "rb") as f: @@ -266,8 +244,7 @@ async def _telegram_send_one_media( if _is_telegram_thread_not_found(media_err) and media_kwargs.get("message_thread_id"): logger.warning( "Thread %s not found for media send, retrying without message_thread_id", - media_kwargs["message_thread_id"], - ) + media_kwargs["message_thread_id"]) f.seek(0) media_kwargs.pop("message_thread_id", None) return await _send(**media_kwargs) @@ -275,8 +252,7 @@ async def _telegram_send_one_media( if media_kwargs.get("parse_mode") and ("parse" in err_text or "caption" in err_text): logger.warning( "Caption parse failed for media send, retrying plain: %s", - _sanitize_error_text(media_err), - ) + _sanitize_error_text(media_err)) f.seek(0) media_kwargs.pop("parse_mode", None) if not has_html and media_kwargs.get("caption"): @@ -286,17 +262,12 @@ async def _telegram_send_one_media( async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False): - """Send via Telegram Bot API (one-shot, no polling needed). - - Markdown is converted to MarkdownV2 via the gateway adapter's - ``format_message`` so bold/links/headers render; a message that already - contains HTML tags skips that and is sent with ``parse_mode='HTML'``. - Parse failures fall back to plain text so the message still delivers. - """ + """One-shot Telegram Bot API send. Markdown -> MarkdownV2 via the adapter's + ``format_message``; text already containing HTML tags is sent as HTML instead. + Parse failures fall back to plain text so the message still delivers.""" try: from telegram.constants import ParseMode - # Auto-detect HTML tags: if present, skip MarkdownV2 and send as HTML. _has_html = bool(re.search(r'<[a-zA-Z/][^>]*>', message)) if _has_html: formatted = message @@ -324,75 +295,55 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No last_msg = None warnings = [] - # MEDIA caption: a single captionable file + short text rides on the - # bubble as its *formatted* caption. Formatting can inflate a raw - # <1024 string past Telegram's 1024 cap, so re-check in UTF-16 units - # and fall back to a separate body message. + # MEDIA caption: a single captionable file + short text rides on the bubble as + # its *formatted* caption. Formatting can inflate a raw <1024 string past + # Telegram's cap, so re-check in UTF-16 units and fall back to a separate body. _tg_caption = None from gateway.platforms.base import BasePlatformAdapter, utf16_len - _cap, _ = _media_caption_split( - message, media_files, max_caption_len=_TELEGRAM_CAPTION_LIMIT - ) + _cap, _ = _media_caption_split(message, media_files, max_caption_len=_TELEGRAM_CAPTION_LIMIT) if _cap is not None and utf16_len(formatted) <= _TELEGRAM_CAPTION_LIMIT: _tg_caption = formatted formatted = "" # suppress the separate text send below if formatted.strip(): - # Chunk *after* formatting, in UTF-16 units: MarkdownV2/HTML - # escaping inflates text, so a raw-<4096 message can exceed the - # limit once formatted and be rejected as "Message is too long". + # Chunk *after* formatting, in UTF-16 units: MarkdownV2/HTML escaping inflates + # text, so a raw-<4096 message can exceed the limit once formatted. for chunk in BasePlatformAdapter.truncate_message(formatted, 4096, len_fn=utf16_len): last_msg = await _telegram_send_text_chunk( - bot, int_chat_id, chunk, send_parse_mode, _has_html, text_kwargs - ) + bot, int_chat_id, chunk, send_parse_mode, _has_html, text_kwargs) for media_path, is_voice in media_files: if not os.path.exists(media_path): warning = f"Media file not found, skipping: {media_path}" logger.warning(warning) warnings.append(warning) - # Caption mode suppressed the text send; if the file it was - # meant to caption is gone, deliver the words on their own. + # Caption mode suppressed the text send; if the file it was meant to + # caption is gone, deliver the words on their own. if _tg_caption is not None and last_msg is None: try: last_msg = await _send_telegram_message_with_retry( bot, chat_id=int_chat_id, text=_tg_caption, - parse_mode=send_parse_mode, **text_kwargs - ) + parse_mode=send_parse_mode, **text_kwargs) _tg_caption = None # delivered — don't re-caption a later file except Exception as _cap_err: logger.warning( "Telegram caption-fallback send failed for missing media: %s", - _sanitize_error_text(_cap_err), - ) + _sanitize_error_text(_cap_err)) continue try: last_msg = await _telegram_send_one_media( bot, int_chat_id, media_path, is_voice, caption=_tg_caption, parse_mode=send_parse_mode, has_html=_has_html, - thread_kwargs=thread_kwargs, force_document=force_document, - ) + thread_kwargs=thread_kwargs, force_document=force_document) except Exception as e: warning = _sanitize_error_text(f"Failed to send media {media_path}: {e}") logger.error(warning) warnings.append(warning) if last_msg is None: - error = "No deliverable text or media remained after processing MEDIA tags" - if warnings: - return {"error": error, "warnings": warnings} - return {"error": error} - - result = { - "success": True, - "platform": "telegram", - "chat_id": chat_id, - "message_id": str(last_msg.message_id), - } - if warnings: - result["warnings"] = warnings - return result + return {"error": _NO_DELIVERABLE, **({"warnings": warnings} if warnings else {})} + return _success("telegram", chat_id, warnings, message_id=str(last_msg.message_id)) except ImportError: return {"error": "python-telegram-bot not installed. Run: pip install python-telegram-bot"} except Exception as e: @@ -408,9 +359,24 @@ def _live_runner(): return None +def _live_adapter(platform, *, lookup_failed_warning=None): + """Return ``(runner, adapter)`` for the running gateway, or ``(runner, None)``. + A runner whose adapter lookup raises is logged when a warning is given, never + silently swallowed (a silent fall-through could recreate a reconnect storm).""" + runner = _live_runner() + if runner is None: + return None, None + try: + return runner, runner.adapters.get(platform) + except Exception: + if lookup_failed_warning: + logger.warning(lookup_failed_warning, exc_info=True) + return runner, None + + def _plugin_standalone_sender(platform_name, *, label=None, discover=True): - """Return ``(standalone_sender_fn, None)`` for a registered platform plugin, - or ``(None, error_dict)``. ``discover`` runs the idempotent plugin scan first.""" + """``(standalone_sender_fn, None)`` for a registered plugin or ``(None, error_dict)``; + ``discover`` runs the idempotent plugin scan first.""" from gateway.platform_registry import platform_registry if discover: @@ -423,8 +389,7 @@ def _plugin_standalone_sender(platform_name, *, label=None, discover=True): async def _registry_standalone_send(platform_name, pconfig, chat_id, message, thread_id=None): - """One-shot text send through a migrated plugin's ``standalone_sender_fn`` - (the former inline ``_send_`` helpers now live in the plugins).""" + """One-shot text send through a plugin's ``standalone_sender_fn``.""" sender, err = _plugin_standalone_sender(platform_name) if err: return err @@ -432,13 +397,10 @@ async def _registry_standalone_send(platform_name, pconfig, chat_id, message, th async def _resolve_slack_user_target(token, chat_id): - """Resolve ``user:U...`` / ``user_name:`` to a D... DM conversation. - - chat.postMessage needs a conversation ID, so user targets go through - conversations.open; ``user_name:`` first maps to a user ID via users.list - (stable handle match only). Other ids pass through unchanged. - Returns ``(chat_id, None)`` or ``(None, error_dict)``. - """ + """Resolve ``user:U...`` / ``user_name:`` to a D... DM conversation + (chat.postMessage needs a conversation ID). ``user_name:`` maps to a user id via + users.list first (stable handle match only); other ids pass through unchanged. + Returns ``(chat_id, None)`` or ``(None, error_dict)``.""" if not (chat_id.startswith("user:") or chat_id.startswith("user_name:")): return chat_id, None try: @@ -447,19 +409,17 @@ async def _resolve_slack_user_target(token, chat_id): return None, {"error": "aiohttp not installed. Run: pip install aiohttp"} try: from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp - _proxy = resolve_proxy_url() - _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(_proxy) - base_url = "https://slack.com/api" + _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(resolve_proxy_url()) headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} async def post_api(session, method, payload): - async with session.post(f"{base_url}/{method}", headers=headers, json=payload, **_req_kw) as resp: + async with session.post( + f"https://slack.com/api/{method}", headers=headers, json=payload, **_req_kw) as resp: return await resp.json() async def resolve_user_name(session, name): query = name.strip().lstrip("@").lower() - matches = [] - cursor = None + matches, cursor = [], None for _page in range(20): payload = {"limit": 200} if cursor: @@ -470,8 +430,7 @@ async def _resolve_slack_user_target(token, chat_id): for member in data.get("members", []): if member.get("deleted") or member.get("is_bot"): continue - # Match the stable handle only: display/real names are - # mutable and non-unique enough to DM the wrong person. + # Stable handle only: display/real names are mutable and non-unique. if str(member.get("name", "")).strip().lower() == query: matches.append(member) cursor = (data.get("response_metadata") or {}).get("next_cursor") @@ -495,8 +454,7 @@ async def _resolve_slack_user_target(token, chat_id): if not opened.get("ok"): return None, _error( f"Slack conversations.open error: {opened.get('error', 'unknown')}. " - "Check bot permissions (im:write)." - ) + "Check bot permissions (im:write).") dm_id = (opened.get("channel") or {}).get("id") if not dm_id: return None, _error("Slack conversations.open did not return a DM channel ID") @@ -505,28 +463,60 @@ async def _resolve_slack_user_target(token, chat_id): return None, _error(f"Slack DM resolution failed: {e}") -async def _send_signal(extra, chat_id, message, media_files=None): - """Send via signal-cli JSON-RPC, text and/or attachments. +async def _signal_send_batch(post, scheduler, rl, idx, n_batches, att_batch, batch_message): + """One Signal batch under the scheduler with rate-limit retries. None on success, + False when retries were exhausted (batch lost), error dict for a non-rate-limit RPC error.""" + n = len(att_batch) + for attempt in range(1, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS + 1): + try: + await scheduler.acquire(n) + _rpc_t0 = time.monotonic() + data = await post(att_batch, batch_message) + _rpc_duration = time.monotonic() - _rpc_t0 + if "error" not in data: + await scheduler.report_rpc_duration(_rpc_duration, n) + return None - Attachments go in batches of SIGNAL_MAX_ATTACHMENTS_PER_MSG metered by the - process-wide SignalAttachmentScheduler — the same bucket the gateway - adapter uses, so tool sends and inbound replies share rate-limit state. - """ + err = data["error"] + if not rl._is_signal_rate_limit_error(err): + return _error(f"Signal RPC error on batch {idx + 1}/{n_batches}: {err}") + + server_retry_after = rl._extract_retry_after_seconds(err) + scheduler.feedback(server_retry_after, n) + retry_after_label = f"{server_retry_after:.0f}s" if server_retry_after else "unknown" + + if attempt >= rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS: + logger.error( + "Signal: rate-limit retries exhausted on batch %d/%d " + "(%d attachments lost, server retry_after=%s)", + idx + 1, n_batches, n, retry_after_label) + return False + logger.warning( + "Signal: rate-limited on batch %d/%d " + "(attempt %d/%d, server retry_after=%s); " + "scheduler will pace the retry", + idx + 1, n_batches, attempt, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, retry_after_label) + except Exception as e: + if attempt >= rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS: + logger.error( + "Signal: send error on batch %d/%d after %d attempts: %s", + idx + 1, n_batches, attempt, str(e)) + return False + logger.warning( + "Signal: transient error on batch %d/%d (attempt %d/%d): %s; will retry", + idx + 1, n_batches, attempt, rl.SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, str(e)) + + +async def _send_signal(extra, chat_id, message, media_files=None): + """signal-cli JSON-RPC send. Attachments go in SIGNAL_MAX_ATTACHMENTS_PER_MSG batches + metered by the process-wide SignalAttachmentScheduler — the same bucket the gateway + adapter uses, so tool sends and inbound replies share rate-limit state.""" try: import httpx except ImportError: return {"error": "httpx not installed"} - from gateway.platforms.signal_rate_limit import ( - SIGNAL_BATCH_PACING_NOTICE_THRESHOLD, - SIGNAL_MAX_ATTACHMENTS_PER_MSG, - SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, - _extract_retry_after_seconds, - _format_wait, - _is_signal_rate_limit_error, - _signal_send_timeout, - get_scheduler, - ) + from gateway.platforms import signal_rate_limit as rl from gateway.platforms.signal_format import markdown_to_signal try: @@ -536,129 +526,74 @@ async def _send_signal(extra, chat_id, message, media_files=None): return {"error": "Signal account not configured"} valid_media = media_files or [] - attachment_paths = [] + attachment_paths = [path for path, _is_voice in valid_media if os.path.exists(path)] for media_path, _is_voice in valid_media: - if os.path.exists(media_path): - attachment_paths.append(media_path) - else: + if not os.path.exists(media_path): logger.warning("Signal media file not found, skipping: %s", media_path) # No attachments still means one (text-only) batch; with attachments # the text rides on batch #0 so it isn't repeated per batch. - if attachment_paths: - att_batches = [ - attachment_paths[i:i + SIGNAL_MAX_ATTACHMENTS_PER_MSG] - for i in range(0, len(attachment_paths), SIGNAL_MAX_ATTACHMENTS_PER_MSG) - ] - else: - att_batches = [[]] + per_batch = rl.SIGNAL_MAX_ATTACHMENTS_PER_MSG + att_batches = [ + attachment_paths[i:i + per_batch] for i in range(0, len(attachment_paths), per_batch) + ] or [[]] + n_batches = len(att_batches) plain_text, text_styles = markdown_to_signal(message) - def _rpc_params(text): + async def _rpc_send(text, *, id_prefix, timeout, attachments=None, styled=False): params = {"account": account, "message": text} if chat_id.startswith("group:"): params["groupId"] = chat_id[6:] else: params["recipient"] = [chat_id] - return params - - async def _rpc_send(params, *, id_prefix, timeout): - payload = { - "jsonrpc": "2.0", - "method": "send", - "params": params, - "id": f"{id_prefix}_{int(time.time() * 1000)}", - } - async with httpx.AsyncClient(timeout=timeout) as client: - return await client.post(f"{http_url}/api/v1/rpc", json=payload) - - async def _post(batch_attachments, batch_message): - params = _rpc_params(batch_message) - if batch_message and text_styles: + if styled and text and text_styles: if len(text_styles) == 1: params["textStyle"] = text_styles[0] else: params["textStyles"] = text_styles - if batch_attachments: - params["attachments"] = batch_attachments - timeout = _signal_send_timeout(len(batch_attachments) if batch_attachments else 0) - resp = await _rpc_send(params, id_prefix="send", timeout=timeout) + if attachments: + params["attachments"] = attachments + payload = { + "jsonrpc": "2.0", + "method": "send", + "params": params, + "id": f"{id_prefix}_{int(time.time() * 1000)}"} + async with httpx.AsyncClient(timeout=timeout) as client: + return await client.post(f"{http_url}/api/v1/rpc", json=payload) + + async def _post(batch_attachments, batch_message): + resp = await _rpc_send( + batch_message, id_prefix="send", timeout=rl._signal_send_timeout(len(batch_attachments)), + attachments=batch_attachments, styled=True) resp.raise_for_status() return resp.json() - async def _send_inline_notice(text: str) -> None: - """Best-effort one-shot RPC for a user-facing pacing notice.""" - try: - await _rpc_send(_rpc_params(text), id_prefix="notice", timeout=30.0) - except Exception as _e: - logger.warning("Signal: inline notice failed: %s", _e) - - scheduler = get_scheduler() + scheduler = rl.get_scheduler() logger.info( "send_message Signal: scheduler state=%s, %d attachment(s) in %d batch(es)", - scheduler.state(), len(attachment_paths), len(att_batches), - ) + scheduler.state(), len(attachment_paths), n_batches) failed_batches: list[int] = [] for idx, att_batch in enumerate(att_batches): n = len(att_batch) if n > 0: estimated = scheduler.estimate_wait(n) - if estimated >= SIGNAL_BATCH_PACING_NOTICE_THRESHOLD: - await _send_inline_notice( - f"(More images coming — pausing ~{_format_wait(estimated)} " - f"for Signal rate limit, batch {idx + 1}/{len(att_batches)}.)" - ) + if estimated >= rl.SIGNAL_BATCH_PACING_NOTICE_THRESHOLD: + # Best-effort one-shot RPC for a user-facing pacing notice. + notice = ( + f"(More images coming — pausing ~{rl._format_wait(estimated)} " + f"for Signal rate limit, batch {idx + 1}/{n_batches}.)") + try: + await _rpc_send(notice, id_prefix="notice", timeout=30.0) + except Exception as _e: + logger.warning("Signal: inline notice failed: %s", _e) - batch_message = plain_text if idx == 0 else "" - - for attempt in range(1, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS + 1): - try: - await scheduler.acquire(n) - _rpc_t0 = time.monotonic() - data = await _post(att_batch, batch_message) - _rpc_duration = time.monotonic() - _rpc_t0 - if "error" not in data: - await scheduler.report_rpc_duration(_rpc_duration, n) - break - - err = data["error"] - - if not _is_signal_rate_limit_error(err): - return _error(f"Signal RPC error on batch {idx + 1}/{len(att_batches)}: {err}") - - server_retry_after = _extract_retry_after_seconds(err) - scheduler.feedback(server_retry_after, n) - - if attempt >= SIGNAL_RATE_LIMIT_MAX_ATTEMPTS: - failed_batches.append(idx + 1) - logger.error( - "Signal: rate-limit retries exhausted on batch %d/%d " - "(%d attachments lost, server retry_after=%s)", - idx + 1, len(att_batches), n, - f"{server_retry_after:.0f}s" if server_retry_after else "unknown", - ) - break - logger.warning( - "Signal: rate-limited on batch %d/%d " - "(attempt %d/%d, server retry_after=%s); " - "scheduler will pace the retry", - idx + 1, len(att_batches), - attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, - f"{server_retry_after:.0f}s" if server_retry_after else "unknown", - ) - except Exception as e: - if attempt >= SIGNAL_RATE_LIMIT_MAX_ATTEMPTS: - failed_batches.append(idx + 1) - logger.error( - "Signal: send error on batch %d/%d after %d attempts: %s", - idx + 1, len(att_batches), attempt, str(e) - ) - break - logger.warning( - "Signal: transient error on batch %d/%d (attempt %d/%d): %s; will retry", - idx + 1, len(att_batches), attempt, SIGNAL_RATE_LIMIT_MAX_ATTEMPTS, str(e) - ) + outcome = await _signal_send_batch( + _post, scheduler, rl, idx, n_batches, att_batch, plain_text if idx == 0 else "") + if outcome is False: + failed_batches.append(idx + 1) + elif outcome is not None: + return outcome warnings = [] if len(attachment_paths) < len(valid_media): @@ -666,56 +601,35 @@ async def _send_signal(extra, chat_id, message, media_files=None): if failed_batches: warnings.append( f"Signal rate-limited {len(failed_batches)} batch(es) " - f"(#{', #'.join(str(b) for b in failed_batches)})" - ) + f"(#{', #'.join(str(b) for b in failed_batches)})") - if failed_batches and len(failed_batches) == len(att_batches): + if failed_batches and len(failed_batches) == n_batches: return _error( - f"Signal: every batch ({len(att_batches)}) hit rate limit; " - f"no attachments delivered" - ) + f"Signal: every batch ({n_batches}) hit rate limit; no attachments delivered") - result = {"success": True, "platform": "signal", "chat_id": _display_chat_id("signal", chat_id)} - if warnings: - result["warnings"] = warnings - return result + return _success("signal", _display_chat_id("signal", chat_id), warnings) except Exception as e: return _error(f"Signal send failed: {e}") async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None, thread_id=None): - """Send via the Matrix adapter so native media uploads are preserved. - - Prefer the live gateway adapter: one persistent olm/megolm session for all - sends. Ephemeral per-send connects re-init E2EE and claim one-time keys, - which under bursts exhausts recipient OTKs and silently drops messages — - so the ephemeral connect/disconnect path is only for standalone/cron. - """ + """Matrix adapter send (native media preserved). Prefer the live gateway adapter's + persistent olm/megolm session: ephemeral per-send connects re-init E2EE and claim + one-time keys, which under bursts exhausts recipient OTKs and silently drops + messages — so the ephemeral connect/disconnect path is only for standalone/cron.""" media_files = media_files or [] metadata = {"thread_id": thread_id} if thread_id else None - # A runner that exists but whose adapter lookup fails is logged, not - # swallowed: a silent fall-through would recreate the reconnect storm. - live_adapter = None - runner = _live_runner() - if runner is not None: - try: - from gateway.config import Platform - live_adapter = runner.adapters.get(Platform.MATRIX) - except Exception: - logger.warning( - "Matrix: live gateway adapter lookup failed; falling back to an " - "ephemeral connect (may re-init E2EE per send)", - exc_info=True, - ) - live_adapter = None - + from gateway.config import Platform + _, live_adapter = _live_adapter( + Platform.MATRIX, + lookup_failed_warning=( + "Matrix: live gateway adapter lookup failed; falling back to an " + "ephemeral connect (may re-init E2EE per send)")) if live_adapter is not None: # Owned by the gateway — must NOT be disconnected; return before the # ephemeral adapter (and its ``finally`` disconnect) exists. - return await _matrix_send_core( - live_adapter, chat_id, message, media_files, metadata - ) + return await _matrix_send_core(live_adapter, chat_id, message, media_files, metadata) # --- Fallback: ephemeral adapter (standalone / cron context) --- try: @@ -725,12 +639,9 @@ async def _send_matrix_via_adapter(pconfig, chat_id, message, media_files=None, adapter = MatrixAdapter(pconfig) try: - connected = await adapter.connect() - if not connected: + if not await adapter.connect(): return _error("Matrix connect failed") - return await _matrix_send_core( - adapter, chat_id, message, media_files, metadata - ) + return await _matrix_send_core(adapter, chat_id, message, media_files, metadata) except Exception as e: return _error(f"Matrix send failed: {e}") finally: @@ -763,19 +674,12 @@ async def _matrix_send_core(adapter, chat_id, message, media_files, metadata): else: method = adapter.send_document last_result = await method(chat_id, media_path, metadata=metadata) - if not last_result.success: return _error(f"Matrix media send failed: {last_result.error}") if last_result is None: - return {"error": "No deliverable text or media remained after processing MEDIA tags"} - - return { - "success": True, - "platform": "matrix", - "chat_id": chat_id, - "message_id": last_result.message_id, - } + return {"error": _NO_DELIVERABLE} + return _success("matrix", chat_id, message_id=last_result.message_id) async def _send_weixin(pconfig, chat_id, message, media_files=None): @@ -789,12 +693,8 @@ async def _send_weixin(pconfig, chat_id, message, media_files=None): try: return await send_weixin_direct( - extra=pconfig.extra, - token=pconfig.token, - chat_id=chat_id, - message=message, - media_files=media_files, - ) + extra=pconfig.extra, token=pconfig.token, chat_id=chat_id, + message=message, media_files=media_files) except Exception as e: return _error(f"Weixin send failed: {e}") @@ -810,16 +710,14 @@ async def _send_bluebubbles(extra, chat_id, message): try: from gateway.config import PlatformConfig - pconfig = PlatformConfig(extra=extra) - adapter = BlueBubblesAdapter(pconfig) - connected = await adapter.connect() - if not connected: + adapter = BlueBubblesAdapter(PlatformConfig(extra=extra)) + if not await adapter.connect(): return _error("BlueBubbles: failed to connect to server") try: result = await adapter.send(chat_id, message) if not result.success: return _error(f"BlueBubbles send failed: {result.error}") - return {"success": True, "platform": "bluebubbles", "chat_id": chat_id, "message_id": result.message_id} + return _success("bluebubbles", chat_id, message_id=result.message_id) finally: await adapter.disconnect() except Exception as e: @@ -833,8 +731,7 @@ async def _send_qqbot(pconfig, chat_id, message): except ImportError: return _error("QQBot direct send requires httpx. Run: pip install httpx") - # Profile-scoped secret lookup so a multiplex profile never borrows - # another profile's QQ credentials. + # Profile-scoped lookup so a multiplex profile never borrows another's QQ credentials. from gateway.config import _getenv extra = pconfig.extra or {} @@ -846,36 +743,28 @@ async def _send_qqbot(pconfig, chat_id, message): try: async with httpx.AsyncClient(timeout=15) as client: - # Step 1: Get access token token_resp = await client.post( "https://bots.qq.com/app/getAppAccessToken", - json={"appId": str(appid), "clientSecret": str(secret)}, - ) + json={"appId": str(appid), "clientSecret": str(secret)}) if token_resp.status_code != 200: return _error(f"QQBot token request failed: {token_resp.status_code}") - token_data = token_resp.json() - access_token = token_data.get("access_token") + access_token = token_resp.json().get("access_token") if not access_token: return _error("QQBot: no access_token in response") - # Step 2: QQ Bot API has separate endpoints for guild channels, C2C - # (private) and groups; try them in that order, first 2xx wins. + # Separate endpoints for guild channels, C2C (private) and groups; first 2xx wins. headers = { - "Authorization": f"QQBot {access_token}", - "Content-Type": "application/json", - } + "Authorization": f"QQBot {access_token}", "Content-Type": "application/json"} payload = {"content": message[:4000], "msg_type": 0} endpoints = ( ("channel", f"https://api.sgroup.qq.com/channels/{chat_id}/messages"), ("c2c", f"https://api.sgroup.qq.com/v2/users/{chat_id}/messages"), - ("group", f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages"), - ) + ("group", f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages")) statuses = [] for kind, url in endpoints: resp = await client.post(url, json=payload, headers=headers) if resp.status_code in {200, 201}: - return {"success": True, "platform": "qqbot", "chat_id": chat_id, - "message_id": resp.json().get("id")} + return _success("qqbot", chat_id, message_id=resp.json().get("id")) statuses.append(f"{kind}={resp.status_code}") return _error(f"QQBot send failed: {' '.join(statuses)}") except Exception as e: @@ -883,8 +772,8 @@ async def _send_qqbot(pconfig, chat_id, message): async def _send_yuanbao(chat_id, message, media_files=None): - """Send via the running Yuanbao adapter's persistent WebSocket (no - throwaway client possible). chat_id: ``group:``, ``direct:`` or ````.""" + """Send via the running Yuanbao adapter's persistent WebSocket (no throwaway client + possible). chat_id: ``group:``, ``direct:`` or ````.""" try: from gateway.platforms.yuanbao import get_active_adapter, send_yuanbao_direct except ImportError: @@ -894,8 +783,7 @@ async def _send_yuanbao(chat_id, message, media_files=None): if adapter is None: return _error( "Yuanbao adapter is not running. " - "Start the gateway with yuanbao platform enabled first." - ) + "Start the gateway with yuanbao platform enabled first.") try: return await send_yuanbao_direct(adapter, chat_id, message, media_files=media_files) diff --git a/tools/send_message_targets.py b/tools/send_message_targets.py index ec2fc5210d..73aa28afb1 100644 --- a/tools/send_message_targets.py +++ b/tools/send_message_targets.py @@ -7,12 +7,10 @@ logger = logging.getLogger("tools.send_message_tool") _TELEGRAM_TOPIC_TARGET_RE = re.compile(r"^\s*(-?\d+)(?::(\d+))?\s*$") _FEISHU_TARGET_RE = re.compile(r"^\s*((?:oc|ou|on|chat|open)_[-A-Za-z0-9]+)(?::([-A-Za-z0-9_]+))?\s*$") -# Slack conversation IDs: C (public channel), G (private/group channel), D (DM). -# Must be uppercase alphanumeric, 9+ chars. User IDs (U...) are parsed as -# explicit user targets (``user:U...``) and are converted to D... conversations -# via conversations.open before chat.postMessage — posting directly to a U/W -# ID fails because the API requires a conversation ID. ``@handle`` targets are -# resolved through users.list first (``user_name:...``). +# Slack conversation IDs: C (public), G (private/group), D (DM); uppercase alnum, 9+ chars. +# User IDs (U...) become ``user:U...`` and are opened as D... conversations before +# chat.postMessage (posting straight to a U/W id fails); ``@handle`` -> ``user_name:...`` +# is resolved through users.list first. _SLACK_TARGET_RE = re.compile(r"^\s*([CGD][A-Z0-9]{8,})\s*$") _SLACK_USER_ID_RE = re.compile(r"^\s*(U[A-Z0-9]{8,})\s*$") _SLACK_USER_NAME_RE = re.compile(r"^\s*@([A-Za-z0-9._-]{1,80})\s*$") @@ -23,47 +21,32 @@ _WEIXIN_TARGET_RE = re.compile(r"^\s*((?:wxid|gh|v\d+|wm|wb)_[A-Za-z0-9_-]+|[A-Z _YUANBAO_TARGET_RE = re.compile(r"^\s*((?:group|direct):[^:]+)\s*$") # Discord snowflake IDs are numeric, same regex pattern as Telegram topic targets. _NUMERIC_TOPIC_RE = _TELEGRAM_TOPIC_TARGET_RE -# Platforms that address recipients by phone number and accept E.164 format -# (with a leading '+'). Without this, "+15551234567" fails the isdigit() check -# below and falls through to channel-name resolution, which has no way to -# resolve a raw phone number. Keeping the '+' preserves the E.164 form that -# downstream adapters (signal, etc.) expect. +# Platforms addressing recipients by E.164 phone number ("+1555..."): without this the +# '+' fails the isdigit() rule and falls through to channel-name resolution, which cannot +# resolve a raw number. The '+' is kept because the adapters expect the E.164 form. _PHONE_PLATFORMS = frozenset({"photon", "signal", "sms", "whatsapp"}) _E164_TARGET_RE = re.compile(r"^\s*\+(\d{7,15})\s*$") # Photon DM chat GUID (mirrors _DM_CHAT_GUID_RE in the photon adapter). _PHOTON_DM_GUID_RE = re.compile(r"^any;-;\+\d{6,}$") -# WhatsApp JIDs: group chats (@g.us), individual users -# (@s.whatsapp.net), linked identities (@lid), and broadcast / -# newsletter chats. These are explicit native targets the bridge accepts -# verbatim — they must NOT fall through to home-channel resolution. +# WhatsApp JIDs (groups @g.us, users @s.whatsapp.net, linked ids @lid, +# broadcast/newsletter): native targets the bridge accepts verbatim — never home-channel. _WHATSAPP_JID_RE = re.compile( - r"^\s*[\w-]+@(?:g\.us|s\.whatsapp\.net|lid|broadcast|newsletter)\s*$", - re.IGNORECASE, -) -# Buzz channels and DMs use native UUID identifiers. They are explicit -# targets and must never substitute the configured home channel. + r"^\s*[\w-]+@(?:g\.us|s\.whatsapp\.net|lid|broadcast|newsletter)\s*$", re.IGNORECASE) +# Buzz channels/DMs are native UUIDs: explicit targets, never the home channel. _BUZZ_UUID_RE = re.compile( - r"^\s*[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\s*$", - re.IGNORECASE, -) -# Email addresses — a valid email like "user@domain.com" should be treated as -# an explicit target for the email platform, not fall through to channel-name -# resolution which has no way to resolve a raw address. + r"^\s*[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\s*$", re.IGNORECASE) +# A valid address is an explicit email target, not a channel name to resolve. _EMAIL_TARGET_RE = re.compile(r"^\s*[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\s*$") -# Most platforms read their home channel from "_HOME_CHANNEL", but a -# few diverge. Email reads EMAIL_HOME_ADDRESS (see gateway/config.py), so the -# generic "_HOME_CHANNEL" hint would point users at a variable that is -# never read. Map the exceptions so the error guidance is actually actionable. +# Home-channel env var exceptions to "_HOME_CHANNEL" (email reads +# EMAIL_HOME_ADDRESS, see gateway/config.py) so the error guidance is actionable. _HOME_CHANNEL_ENV_OVERRIDES = {"email": "EMAIL_HOME_ADDRESS"} - _UNRESOLVED = object() # sentinel: stop parsing, target is NOT explicit (skip generic rules) -# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id), -# None to fall through to the generic rules in _parse_target_ref, or -# _UNRESOLVED. Order inside each parser matters (e.g. Slack thread form -# before bare conversation id). +# Per-platform explicit-target parsers: target_ref -> (chat_id, thread_id), None to fall +# through to the generic rules in _parse_target_ref, or _UNRESOLVED. Order inside each +# parser matters (e.g. Slack thread form before bare conversation id). def _parse_regex_groups(regex, *, thread_group=True): """Explicit when ``regex`` fully matches: chat_id = group 1, thread = group 2 (or None).""" def parse(ref): @@ -74,6 +57,13 @@ def _parse_regex_groups(regex, *, thread_group=True): return parse +def _parse_regex_stripped(regex): + """Explicit when ``regex`` fully matches; returns the stripped ref verbatim.""" + def parse(ref): + return (ref.strip(), None) if regex.fullmatch(ref) else None + return parse + + def _parse_telegram(ref): # "[:]" or an @username (usernames must not be force-int'd). parsed = _parse_regex_groups(_TELEGRAM_TOPIC_TARGET_RE)(ref) @@ -85,24 +75,26 @@ def _parse_telegram(ref): return (username, None) if username else None +# (regex, chat_id template, thread comes from group 2) — thread form before bare id. +_SLACK_FORMS = ( + (_SLACK_THREAD_TARGET_RE, "{}", True), + (_SLACK_TARGET_RE, "{}", False), + (_SLACK_USER_ID_RE, "user:{}", False), + (_SLACK_MENTION_RE, "user:{}", False), + (_SLACK_USER_NAME_RE, "user_name:{}", False)) + + def _parse_slack(ref): - match = _SLACK_THREAD_TARGET_RE.fullmatch(ref) - if match: - return match.group(1), match.group(2) - match = _SLACK_TARGET_RE.fullmatch(ref) - if match: - return match.group(1), None - match = _SLACK_USER_ID_RE.fullmatch(ref) or _SLACK_MENTION_RE.fullmatch(ref) - if match: - return f"user:{match.group(1)}", None - match = _SLACK_USER_NAME_RE.fullmatch(ref) - return (f"user_name:{match.group(1)}", None) if match else None + for regex, template, has_thread in _SLACK_FORMS: + match = regex.fullmatch(ref) + if match: + return template.format(match.group(1)), (match.group(2) if has_thread else None) + return None def _parse_matrix(ref): - # ":$" addresses a thread (rfind: room ids contain ':'). - # Bare "!room" / "@user" are explicit too, but via the generic rule so the - # numeric check keeps precedence exactly as before. + # ":$" addresses a thread (rfind: room ids contain ':'). Bare "!room" / + # "@user" are explicit too, but via the generic rule so the numeric check keeps precedence. trimmed = ref.strip() split_idx = trimmed.rfind(":$") if split_idx > 0: @@ -110,16 +102,9 @@ def _parse_matrix(ref): return None -def _parse_regex_stripped(regex): - """Explicit when ``regex`` fully matches; returns the stripped ref verbatim.""" - def parse(ref): - return (ref.strip(), None) if regex.fullmatch(ref) else None - return parse - - def _parse_yuanbao(ref): - # "group:" / "direct:"; a bare number is a group code, never a - # generic numeric chat id (yuanbao never falls through to the generic rules). + # "group:" / "direct:"; a bare number is a group code, never a generic + # numeric chat id (yuanbao never falls through to the generic rules). match = _YUANBAO_TARGET_RE.fullmatch(ref) if match: return match.group(1), None @@ -129,8 +114,8 @@ def _parse_yuanbao(ref): def _parse_nonempty(ref): - # ntfy topics and WeCom ids (wr/wc groups, wo/bare users — the adapter - # picks the send command) are explicit whenever non-empty. + # ntfy topics and WeCom ids (wr/wc groups, wo/bare users — the adapter picks the + # send command) are explicit whenever non-empty. stripped = ref.strip() return (stripped, None) if stripped else None @@ -160,18 +145,15 @@ _PLATFORM_PARSERS = { "signal": _parse_signal, "wecom": _parse_nonempty, # Photon DM GUIDs are adapter-native ids (mirrors the react handler). - "photon": _parse_regex_stripped(_PHOTON_DM_GUID_RE), -} + "photon": _parse_regex_stripped(_PHOTON_DM_GUID_RE)} def _parse_target_ref(platform_name: str, target_ref: str): """Parse a tool target into (chat_id, thread_id, explicit). - Platform parser first, then the shared rules in this order: E.164 phone - numbers (keeping the '+' the signal/sms/whatsapp adapters expect), bare - numeric ids, Matrix ``!room``/``@user``, XMPP JIDs (user@server or - room@conference.server). Anything else is not explicit and goes to - channel-directory resolution. + Platform parser first, then the shared rules in this order: E.164 phone numbers + (keeping the '+' the signal/sms/whatsapp adapters expect), bare numeric ids, Matrix + ``!room``/``@user``, XMPP JIDs. Anything else goes to channel-directory resolution. """ parser = _PLATFORM_PARSERS.get(platform_name) if parser is not None: @@ -196,22 +178,18 @@ def resolve_send_target( ) -> tuple[str | None, str | None, str | None]: """Resolve one send target the same way for every caller (model tool, CLI, cron). - Channel-directory IDs are trusted. Plugin platforms must explicitly parse - native target syntax; for the model-facing send tool (the default), a - target that can't be resolved is an error — the model can read the error - and pick a listed target instead. + Channel-directory IDs are trusted. Plugin platforms must explicitly parse native + target syntax; for the model-facing send tool (the default) an unresolvable target + is an error the model can read and pick a listed target instead. - ``pass_unresolved_references=True`` restores the old pass-through behavior for - callers that have no model in the loop (cron delivering a stored job's - output, react/unreact on platform-native message ids): if the target - can't be resolved and the platform is built in, or is a plugin platform - that declares no parser, the string is handed to the adapter exactly as - written and the adapter decides whether it's valid. A plugin platform - that DOES declare a parser stays strict for every caller — its parser is - the authority on native syntax. + ``pass_unresolved_references=True`` is for callers with no model in the loop (cron + delivering a stored job's output, react/unreact on platform-native message ids): an + unresolvable target on a built-in platform, or on a plugin platform that declares no + parser, is handed to the adapter exactly as written and the adapter decides. A plugin + platform that DOES declare a parser stays strict for every caller. - The optional validator has the final say over parser-normalized, - directory-resolved, and passed-through IDs alike. + The optional validator has the final say over parser-normalized, directory-resolved, + and passed-through IDs alike. """ from gateway.config import Platform from gateway.platform_registry import platform_registry @@ -224,9 +202,7 @@ def resolve_send_target( try: verdict = entry.validate_target_ref_fn(candidate) except Exception: - logger.debug( - "Plugin target validator failed for %s", platform_name, exc_info=True - ) + logger.debug("Plugin target validator failed for %s", platform_name, exc_info=True) return f"Target validator failed for platform '{platform_name}'" if verdict is True: return None @@ -234,13 +210,15 @@ def resolve_send_target( return f"Invalid target '{target_ref}' on {platform_name}: {verdict}" return f"Invalid target '{target_ref}' on {platform_name}" + def _validated(chat_id, thread_id): + error = _validate(chat_id) + return (None, None, error) if error else (chat_id, thread_id, None) + if entry is not None and entry.parse_target_ref_fn is not None: try: parsed = entry.parse_target_ref_fn(target_ref) except Exception: - logger.debug( - "Plugin target parser failed for %s", platform_name, exc_info=True - ) + logger.debug("Plugin target parser failed for %s", platform_name, exc_info=True) return None, None, f"Target parser failed for platform '{platform_name}'" if parsed is not None: if ( @@ -248,31 +226,15 @@ def resolve_send_target( or len(parsed) != 2 or not isinstance(parsed[0], str) or not parsed[0] - or (parsed[1] is not None and not isinstance(parsed[1], str)) - ): + or (parsed[1] is not None and not isinstance(parsed[1], str))): return ( - None, - None, - f"Target parser for platform '{platform_name}' returned an invalid result", - ) - parsed_chat_id, parsed_thread_id = parsed - error = _validate(parsed_chat_id) - return (None, None, error) if error else ( - parsed_chat_id, - parsed_thread_id, - None, - ) + None, None, + f"Target parser for platform '{platform_name}' returned an invalid result") + return _validated(*parsed) - parsed_chat_id, parsed_thread_id, explicit = _parse_target_ref( - platform_name, target_ref - ) + parsed_chat_id, parsed_thread_id, explicit = _parse_target_ref(platform_name, target_ref) if explicit and parsed_chat_id is not None: - error = _validate(parsed_chat_id) - return (None, None, error) if error else ( - parsed_chat_id, - parsed_thread_id, - None, - ) + return _validated(parsed_chat_id, parsed_thread_id) resolution_failed = False try: @@ -283,16 +245,8 @@ def resolve_send_target( resolved = None resolution_failed = True if resolved: - parsed_chat_id, parsed_thread_id, _ = _parse_target_ref( - platform_name, resolved - ) - chat_id = parsed_chat_id or resolved - error = _validate(chat_id) - return (None, None, error) if error else ( - chat_id, - parsed_thread_id, - None, - ) + parsed_chat_id, parsed_thread_id, _ = _parse_target_ref(platform_name, resolved) + return _validated(parsed_chat_id or resolved, parsed_thread_id) is_builtin = platform_name in {member.value for member in Platform} if entry is None and not is_builtin: @@ -306,24 +260,20 @@ def resolve_send_target( logger.debug( "Handing unresolved target '%s' to the %s adapter unchanged " "(the adapter validates it)", - target_ref, platform_name, - ) + target_ref, platform_name) return target_ref, None, None if entry is not None and entry.source == "plugin" and not is_builtin: if pass_unresolved_references and entry.parse_target_ref_fn is None: return _pass_through_unresolved() return ( - None, - None, + None, None, f"Could not resolve '{target_ref}' on {platform_name}. " - "The plugin parser did not recognize it and no channel-directory entry matched.", - ) + "The plugin parser did not recognize it and no channel-directory entry matched.") if pass_unresolved_references: return _pass_through_unresolved() hint = ( "Try using a numeric channel ID instead." if resolution_failed - else "Use send_message(action='list') to see available targets." - ) + else "Use send_message(action='list') to see available targets.") return None, None, f"Could not resolve '{target_ref}' on {platform_name}. {hint}" diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index 2d08eca521..209fee15d6 100644 --- a/tools/send_message_tool.py +++ b/tools/send_message_tool.py @@ -1,7 +1,7 @@ """Send Message Tool -- cross-channel messaging via platform APIs. Sends a message to a user or channel on any connected messaging platform -(Telegram, Discord, Slack). Supports listing available targets and resolving +(Telegram, Discord, Slack, ...), lists available targets and resolves human-friendly channel names to IDs. Works in both CLI and gateway contexts. """ @@ -14,58 +14,22 @@ from agent.secret_scope import get_secret logger = logging.getLogger(__name__) +# Re-exported: tests and sibling modules import these via tools.send_message_tool. from tools.send_message_targets import ( # noqa: F401 - _BUZZ_UUID_RE, - _E164_TARGET_RE, - _EMAIL_TARGET_RE, - _FEISHU_TARGET_RE, - _HOME_CHANNEL_ENV_OVERRIDES, - _NUMERIC_TOPIC_RE, - _PHONE_PLATFORMS, - _PHOTON_DM_GUID_RE, - _SLACK_MENTION_RE, - _SLACK_TARGET_RE, - _SLACK_THREAD_TARGET_RE, - _SLACK_USER_ID_RE, - _SLACK_USER_NAME_RE, - _TELEGRAM_TOPIC_TARGET_RE, - _WEIXIN_TARGET_RE, - _WHATSAPP_JID_RE, - _YUANBAO_TARGET_RE, - _parse_target_ref, - resolve_send_target, -) + _HOME_CHANNEL_ENV_OVERRIDES, _SLACK_USER_ID_RE, _parse_target_ref, resolve_send_target) from tools.send_message_senders import ( # noqa: F401 - _AUDIO_EXTS, - _CAPTIONABLE_EXTS, - _DEFAULT_CAPTION_LIMIT, - _GENERIC_SECRET_ASSIGN_RE, - _IMAGE_EXTS, - _TELEGRAM_CAPTION_LIMIT, - _TELEGRAM_SEND_AUDIO_EXTS, - _URL_SECRET_QUERY_RE, - _VIDEO_EXTS, - _VOICE_EXTS, - _display_chat_id, - _error, - _is_telegram_thread_not_found, - _live_runner, - _matrix_send_core, - _media_caption_split, - _plugin_standalone_sender, - _registry_standalone_send, - _resolve_slack_user_target, - _sanitize_error_text, - _send_bluebubbles, - _send_matrix_via_adapter, - _send_qqbot, - _send_signal, - _send_telegram, - _send_telegram_message_with_retry, - _send_weixin, - _send_yuanbao, - _telegram_retry_delay, -) + _AUDIO_EXTS, _DEFAULT_CAPTION_LIMIT, _IMAGE_EXTS, _TELEGRAM_CAPTION_LIMIT, _VIDEO_EXTS, + _VOICE_EXTS, _error, _live_adapter, _media_caption_split, _plugin_standalone_sender, + _registry_standalone_send, _resolve_slack_user_target, _sanitize_error_text, + _send_bluebubbles, _send_matrix_via_adapter, _send_qqbot, _send_signal, _send_telegram, + _send_weixin, _send_yuanbao) +from tools.registry import tool_error + +# NOTE: ``send_message`` is intentionally NOT registered as an agent-callable model +# tool (the agent must not fire cross-platform messages on its own). The send engine +# here is the shared transport for cron delivery, the ``hermes send`` CLI, the gateway +# kanban notifier and the opt-in MCP server, which import the helpers directly. + def prepare_send_message_platforms() -> None: """Load enabled standalone plugins before tool schemas/cache keys are built.""" @@ -90,10 +54,16 @@ def _split_target(target: str): return parts[0].strip().lower(), (parts[1].strip() if len(parts) > 1 else None) -def _live_adapter(platform): - """Return the running gateway's adapter for ``platform``, or None.""" - runner = _live_runner() - return runner.adapters.get(platform) if runner is not None else None +def _resolve_tool_target(target: str, *, pass_unresolved_references: bool = False): + """Split and resolve a tool ``target``; returns ``(platform_name, chat_id, thread_id, error)``. + ``chat_id`` is None when no ref was given (caller falls back to the home channel).""" + platform_name, target_ref = _split_target(target) + prepare_send_message_platforms() + if not target_ref: + return platform_name, None, None, None + chat_id, thread_id, resolution_error = resolve_send_target( + platform_name, target_ref, pass_unresolved_references=pass_unresolved_references) + return platform_name, chat_id, thread_id, resolution_error def _handle_list(): @@ -106,32 +76,23 @@ def _handle_list(): def _handle_react(args, remove=False): - """Attach (or with ``remove=True`` retract) an emoji reaction via a live - gateway adapter exposing ``add_reaction(chat_id, emoji, message_id)`` / - ``remove_reaction(chat_id, message_id)`` (e.g. photon/iMessage tapbacks). - No standalone fallback: reacting needs the adapter's live message-id state. - """ + """Attach (or with ``remove=True`` retract) an emoji reaction via a live gateway + adapter exposing ``add_reaction(chat_id, emoji, message_id)`` / + ``remove_reaction(chat_id, message_id)`` (e.g. photon/iMessage tapbacks). No + standalone fallback: reacting needs the adapter's live message-id state.""" target = args.get("target", "") emoji = (args.get("emoji") or "").strip() message_id = (args.get("message_id") or "").strip() or None if not target or (not remove and not emoji): - return tool_error( - "Both 'target' and 'emoji' are required when action='react'" - if not remove - else "'target' is required when action='unreact'" - ) + return tool_error("'target' is required when action='unreact'" if remove + else "Both 'target' and 'emoji' are required when action='react'") - platform_name, target_ref = _split_target(target) - chat_id = None - prepare_send_message_platforms() - if target_ref: - # Platform-native ids (e.g. photon GUIDs) match no parser/directory - # entry; hand them to the adapter unchanged and let it validate. - chat_id, _thread_id, resolution_error = resolve_send_target( - platform_name, target_ref, pass_unresolved_references=True - ) - if resolution_error: - return tool_error(resolution_error) + # Platform-native ids (e.g. photon GUIDs) match no parser/directory entry; hand + # them to the adapter unchanged and let it validate. + platform_name, chat_id, _thread_id, resolution_error = _resolve_tool_target( + target, pass_unresolved_references=True) + if resolution_error: + return tool_error(resolution_error) try: from gateway.config import Platform, load_gateway_config @@ -147,21 +108,17 @@ def _handle_react(args, remove=False): if not home: return tool_error( f"No chat specified and no home channel set for {platform_name}. " - f"Use '{platform_name}:chat_id'." - ) + f"Use '{platform_name}:chat_id'.") chat_id = home.chat_id - adapter = _live_adapter(platform) + _, adapter = _live_adapter(platform) if adapter is None: return tool_error( f"Reactions require a live {platform_name} adapter in the running " - "gateway (not available from cron/standalone contexts)." - ) + "gateway (not available from cron/standalone contexts).") react_fn = getattr(adapter, "remove_reaction" if remove else "add_reaction", None) if not callable(react_fn): - return tool_error( - f"Platform '{platform_name}' does not support message reactions." - ) + return tool_error(f"Platform '{platform_name}' does not support message reactions.") kwargs = {"chat_id": chat_id, "message_id": message_id} if not remove: @@ -171,9 +128,7 @@ def _handle_react(args, remove=False): result = _run_async(react_fn(**kwargs)) except Exception as e: return json.dumps(_error(f"Reaction failed: {e}")) - if isinstance(result, dict): - return json.dumps(result) - return json.dumps({"success": bool(result)}) + return json.dumps(result if isinstance(result, dict) else {"success": bool(result)}) def _handle_send(args): @@ -183,17 +138,9 @@ def _handle_send(args): if not target or not message: return tool_error("Both 'target' and 'message' are required when action='send'") - platform_name, target_ref = _split_target(target) - chat_id = None - thread_id = None - - prepare_send_message_platforms() - if target_ref: - chat_id, thread_id, resolution_error = resolve_send_target( - platform_name, target_ref - ) - if resolution_error: - return tool_error(resolution_error) + platform_name, chat_id, thread_id, resolution_error = _resolve_tool_target(target) + if resolution_error: + return tool_error(resolution_error) from tools.interrupt import is_interrupted if is_interrupted(): @@ -211,9 +158,9 @@ def _handle_send(args): from gateway.platforms.base import BasePlatformAdapter - # Capture [[as_document]] before extract_media strips it: image files then - # go through send_document so the original bytes survive (Telegram's - # sendPhoto recompresses to 1280px). + # Capture [[as_document]] before extract_media strips it: image files then go + # through send_document so the original bytes survive (Telegram's sendPhoto + # recompresses to 1280px). force_document_attachments = "[[as_document]]" in message media_files, cleaned_message = BasePlatformAdapter.extract_media(message) @@ -237,18 +184,13 @@ def _handle_send(args): try: from model_tools import _run_async - send_kwargs = { - "thread_id": thread_id, - "media_files": media_files, - "force_document": force_document_attachments, - } + send_kwargs = {"thread_id": thread_id, "media_files": media_files, + "force_document": force_document_attachments} # Only custom plugin handlers receive the complete typed request; the # built-in call contract stays exact. if entry is not None and entry.send_message_handler is not None: send_kwargs["args"] = args - result = _run_async( - _send_to_platform(platform, pconfig, chat_id, cleaned_message, **send_kwargs) - ) + result = _run_async(_send_to_platform(platform, pconfig, chat_id, cleaned_message, **send_kwargs)) if isinstance(result, dict) and result.get("success"): if used_home_channel: result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})" @@ -265,9 +207,9 @@ def _handle_send(args): def _resolve_platform_config(platform_name, config): """Return ``(platform, pconfig, registry_entry, error)`` for a send. - Plugin platforms must be registered; disabled/missing platforms error, - except Weixin, which may be configured purely via .env (synthesized - pconfig so send_message and cron delivery work without a gateway.yaml entry). + Plugin platforms must be registered; disabled/missing platforms error, except + Weixin, which may be configured purely via .env (synthesized pconfig so + send_message and cron delivery work without a gateway.yaml entry). """ from gateway.config import Platform from gateway.platform_registry import platform_registry @@ -287,8 +229,7 @@ def _resolve_platform_config(platform_name, config): if pconfig is None: return None, None, None, ( f"Platform '{platform_name}' is not configured. Set up credentials in " - "~/.hermes/config.yaml or environment variables." - ) + "~/.hermes/config.yaml or environment variables.") return platform, pconfig, entry, None @@ -303,21 +244,17 @@ def _home_chat_id(config, platform, platform_name): home = HomeChannel(platform=platform, chat_id=wx_home, name="Weixin Home") if home: return home.chat_id, None - home_env = _HOME_CHANNEL_ENV_OVERRIDES.get( - platform_name, f"{platform_name.upper()}_HOME_CHANNEL" - ) + home_env = _HOME_CHANNEL_ENV_OVERRIDES.get(platform_name, f"{platform_name.upper()}_HOME_CHANNEL") return None, ( f"No home channel set for {platform_name} to determine where to send the message. " f"Either specify a channel directly with '{platform_name}:CHANNEL_NAME', " - f"or set a home channel via: hermes config set {home_env} " - ) + f"or set a home channel via: hermes config set {home_env} ") def _slack_dm_chat_id(pconfig, chat_id): - """Open Slack user targets as DM conversations: ``user:U...`` / - ``user_name:@handle`` from the parser, or a bare U... id from session - metadata / home-channel config. chat.postMessage needs a conversation ID. - Returns ``(chat_id, None)`` or ``(None, error_dict)``.""" + """Open Slack user targets as DM conversations: ``user:U...`` / ``user_name:@handle`` + from the parser, or a bare U... id from session metadata / home-channel config. + chat.postMessage needs a conversation ID. Returns ``(chat_id, None)`` or ``(None, error_dict)``.""" dm_target = chat_id if dm_target.startswith("U") and _SLACK_USER_ID_RE.fullmatch(dm_target): dm_target = f"user:{dm_target}" @@ -333,13 +270,10 @@ def _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id): from gateway.mirror import mirror_to_session from gateway.session_context import get_session_env return bool(mirror_to_session( - platform_name, - chat_id, - mirror_text, + platform_name, chat_id, mirror_text, source_label=get_session_env("HERMES_SESSION_PLATFORM", "cli"), thread_id=thread_id, - user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None, - )) + user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None)) except Exception: return False @@ -351,15 +285,10 @@ def _weixin_env_pconfig(): if not (wx_token and wx_account): return None from gateway.config import PlatformConfig - return PlatformConfig( - enabled=True, - token=wx_token, - extra={ - "account_id": wx_account, - "base_url": get_secret("WEIXIN_BASE_URL", "").strip(), - "cdn_base_url": get_secret("WEIXIN_CDN_BASE_URL", "").strip(), - }, - ) + return PlatformConfig(enabled=True, token=wx_token, extra={ + "account_id": wx_account, + "base_url": get_secret("WEIXIN_BASE_URL", "").strip(), + "cdn_base_url": get_secret("WEIXIN_CDN_BASE_URL", "").strip()}) def _describe_media_for_mirror(media_files): @@ -378,165 +307,110 @@ def _describe_media_for_mirror(media_files): return "[Sent document attachment]" -def _get_cron_auto_delivery_target(): - """Return the cron scheduler's auto-delivery target for the current run, if any.""" - from gateway.session_context import get_session_env - platform = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower() - chat_id = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip() - if not platform or not chat_id: - return None - thread_id = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None - return {"platform": platform, "chat_id": chat_id, "thread_id": thread_id} - - def _maybe_skip_cron_duplicate_send(platform_name: str, chat_id: str, thread_id: str | None): """Skip redundant cron send_message calls when the scheduler will auto-deliver there.""" - auto_target = _get_cron_auto_delivery_target() - if not auto_target or not ( - auto_target["platform"] == platform_name - and str(auto_target["chat_id"]) == str(chat_id) - and auto_target.get("thread_id") == thread_id - ): + from gateway.session_context import get_session_env + auto_platform = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM", "").strip().lower() + auto_chat_id = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID", "").strip() + if not auto_platform or not auto_chat_id: + return None + auto_thread_id = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None + if not (auto_platform == platform_name and auto_chat_id == str(chat_id) and auto_thread_id == thread_id): return None target_label = f"{platform_name}:{chat_id}" if thread_id is not None: target_label += f":{thread_id}" - return { - "success": True, - "skipped": True, - "reason": "cron_auto_delivery_duplicate_target", + "success": True, "skipped": True, "reason": "cron_auto_delivery_duplicate_target", "target": target_label, "note": ( f"Skipped send_message to {target_label}. This cron job will already auto-deliver " "its final response to that same target. Put the intended user-facing content in " "your final response instead, or use a different target if you want an additional message." - ), - } + )} def _bounded_send_error(detail, max_chars=900): """Bound untrusted adapter/plugin error detail returned by send_message.""" text = str(detail or "send failed") - if len(text) <= max_chars: - return text - return f"{text[: max_chars - 3]}..." + return text if len(text) <= max_chars else f"{text[: max_chars - 3]}..." + + +def _live_media_method(ext, is_voice, force_document): + """Adapter media method name and kind for one live-adapter media file.""" + if force_document: + return "send_document", "document" + if ext in _IMAGE_EXTS: + return "send_image_file", "image" + if ext in _VIDEO_EXTS: + return "send_video", "video" + if is_voice or ext in _AUDIO_EXTS: + return "send_voice", "audio" + return "send_document", "document" async def _send_live_adapter_media( - adapter, - chat_id, - message, - media_files, - *, - thread_id=None, - metadata=None, - force_document=False, -): + adapter, chat_id, message, media_files, *, thread_id=None, metadata=None, force_document=False): """Deliver text and every media descriptor through adapter media APIs. Adapters that only inherit the BasePlatformAdapter stub for a media kind are treated as unsupported rather than silently no-op'd.""" caption, separate_text = _media_caption_split( - message, media_files, max_caption_len=_DEFAULT_CAPTION_LIMIT - ) + message, media_files, max_caption_len=_DEFAULT_CAPTION_LIMIT) last_result = None if separate_text and separate_text.strip(): - last_result = await adapter.send( - chat_id=chat_id, content=separate_text, metadata=metadata - ) + last_result = await adapter.send(chat_id=chat_id, content=separate_text, metadata=metadata) if not last_result.success: return {"error": f"Adapter send failed: {_bounded_send_error(last_result.error)}"} + from gateway.platforms.base import BasePlatformAdapter + total = len(media_files) for index, descriptor in enumerate(media_files): + invalid = {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"} if not isinstance(descriptor, (list, tuple)) or not descriptor: - return {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"} + return invalid media_path = descriptor[0] is_voice = bool(descriptor[1]) if len(descriptor) > 1 else False if not isinstance(media_path, str) or not media_path: - return {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"} + return invalid if not os.path.exists(media_path): return {"error": f"Adapter media send failed: media file {index + 1}/{total} was not found"} ext = os.path.splitext(media_path)[1].lower() - kwargs = { - "caption": caption if index == 0 else None, - "reply_to": thread_id, - "metadata": metadata, - } - if force_document: - method_name, media_kind = "send_document", "document" - elif ext in _IMAGE_EXTS: - method_name, media_kind = "send_image_file", "image" - elif ext in _VIDEO_EXTS: - method_name, media_kind = "send_video", "video" - elif is_voice or ext in _AUDIO_EXTS: - method_name, media_kind = "send_voice", "audio" - else: - method_name, media_kind = "send_document", "document" - - from gateway.platforms.base import BasePlatformAdapter - + method_name, media_kind = _live_media_method(ext, is_voice, force_document) adapter_method = getattr(type(adapter), method_name, None) - base_fallback = getattr(BasePlatformAdapter, method_name) - if adapter_method is None or adapter_method is base_fallback: + if adapter_method is None or adapter_method is getattr(BasePlatformAdapter, method_name): return { "error": ( f"Live adapter does not implement native {media_kind} delivery; " - f"media file {index + 1}/{total} was not sent" - ) - } + f"media file {index + 1}/{total} was not sent")} + kwargs = {"caption": caption if index == 0 else None, "reply_to": thread_id, "metadata": metadata} try: last_result = await getattr(adapter, method_name)(chat_id, media_path, **kwargs) except asyncio.CancelledError: raise except Exception as exc: - return { - "error": ( - f"Adapter media send failed after {index}/{total} files: " - f"{_bounded_send_error(exc)}" - ) - } - if not last_result.success: + detail = _bounded_send_error(exc) + else: + if last_result.success: + continue detail = _bounded_send_error(last_result.error or "media send failed") - return { - "error": f"Adapter media send failed after {index}/{total} files: {detail}" - } + return {"error": f"Adapter media send failed after {index}/{total} files: {detail}"} if last_result is None: return {"error": "No deliverable text or media remained after processing MEDIA tags"} - return { - "success": True, - "message_id": last_result.message_id, - "media_delivered": True, - } + return {"success": True, "message_id": last_result.message_id, "media_delivered": True} async def _send_via_adapter( - platform, - pconfig, - chat_id, - chunk, - *, - thread_id=None, - media_files=None, - force_document=False, -): - """Send via the live in-process gateway adapter (``_gateway_runner_ref``), - else the plugin's ``standalone_sender_fn`` (gateway not in this process, - e.g. cron — the runner weakref is None), else a descriptive error naming - both options. Media descriptors go through the adapter's native media - APIs under the same cross-loop rules as text. - """ + platform, pconfig, chat_id, chunk, *, thread_id=None, media_files=None, force_document=False): + """Send via the live in-process gateway adapter (``_gateway_runner_ref``), else the + plugin's ``standalone_sender_fn`` (gateway not in this process, e.g. cron — the runner + weakref is None), else a descriptive error naming both options. Media descriptors go + through the adapter's native media APIs under the same cross-loop rules as text.""" platform_name = platform.value if hasattr(platform, "value") else str(platform) - runner = _live_runner() - adapter = None - if runner is not None: - try: - adapter = runner.adapters.get(platform) - except Exception: - adapter = None + runner, adapter = _live_adapter(platform) if adapter is not None: try: metadata = {} @@ -545,9 +419,9 @@ async def _send_via_adapter( if platform_name == "ntfy" and chat_id: metadata["publish_topic"] = chat_id metadata = metadata or None - # adapter.send() uses queues/tasks bound to the gateway's loop. - # Awaiting it from another loop (the tool worker thread) deadlocks - # on a cross-loop Future, so dispatch onto the gateway loop instead. + # adapter.send() uses queues/tasks bound to the gateway's loop. Awaiting it + # from another loop (the tool worker thread) deadlocks on a cross-loop + # Future, so dispatch onto the gateway loop instead. gateway_loop = getattr(runner, "_gateway_loop", None) try: current_loop = asyncio.get_running_loop() @@ -562,28 +436,23 @@ async def _send_via_adapter( return {"error": "Gateway loop is not running; cannot dispatch adapter send"} from agent.async_utils import safe_schedule_threadsafe fut = safe_schedule_threadsafe( - make_coro(), gateway_loop, logger=logger, log_message=log_message - ) + make_coro(), gateway_loop, logger=logger, log_message=log_message) if fut is None: return {"error": "Gateway loop unavailable for send dispatch"} # shield: a cancelled caller (agent interrupt) must not cancel the - # already-enqueued gateway send, or a retry would duplicate it. - # No timeout here — the adapter's request timeout and the outer - # _run_async timeout bound the wait. + # already-enqueued gateway send, or a retry would duplicate it. No timeout + # here — the adapter's request timeout and the outer _run_async bound the wait. return await asyncio.shield(asyncio.wrap_future(fut)) if media_files: return await _dispatch( lambda: _send_live_adapter_media( adapter, chat_id, chunk, media_files, - thread_id=thread_id, metadata=metadata, force_document=force_document, - ), - "send_message: failed to schedule media send on gateway loop", - ) + thread_id=thread_id, metadata=metadata, force_document=force_document), + "send_message: failed to schedule media send on gateway loop") result = await _dispatch( lambda: adapter.send(chat_id=chat_id, content=chunk, metadata=metadata), - "send_message: failed to schedule on gateway loop", - ) + "send_message: failed to schedule on gateway loop") except asyncio.CancelledError: raise except Exception as e: @@ -599,43 +468,29 @@ async def _send_via_adapter( entry = platform_registry.get(platform_name) except Exception: entry = None - - if entry is not None and entry.standalone_sender_fn is not None: - try: - result = await entry.standalone_sender_fn( - pconfig, - chat_id, - chunk, - thread_id=thread_id, - media_files=media_files, - force_document=force_document, - ) - except asyncio.CancelledError: - raise - except Exception as e: - logger.debug("Plugin standalone send for %s raised", platform_name, exc_info=True) - return {"error": f"Plugin standalone send failed: {_bounded_send_error(e)}"} - - if isinstance(result, dict) and (result.get("success") or result.get("error")): - if result.get("error"): - return {**result, "error": _bounded_send_error(result["error"])} - return result - return { - "error": ( - f"Plugin standalone send for '{platform_name}' returned an " - f"invalid result: expected a dict with 'success' or 'error' " - f"keys, got {type(result).__name__}" - ) - } - - return { - "error": ( + if entry is None or entry.standalone_sender_fn is None: + return {"error": ( f"No live adapter for platform '{platform_name}'. Is the gateway " f"running with this platform connected? For out-of-process delivery " f"(e.g. cron in a separate process), the platform plugin must " - f"register a standalone_sender_fn on its PlatformEntry." - ) - } + f"register a standalone_sender_fn on its PlatformEntry.")} + try: + result = await entry.standalone_sender_fn( + pconfig, chat_id, chunk, + thread_id=thread_id, media_files=media_files, force_document=force_document) + except asyncio.CancelledError: + raise + except Exception as e: + logger.debug("Plugin standalone send for %s raised", platform_name, exc_info=True) + return {"error": f"Plugin standalone send failed: {_bounded_send_error(e)}"} + if isinstance(result, dict) and (result.get("success") or result.get("error")): + if result.get("error"): + return {**result, "error": _bounded_send_error(result["error"])} + return result + return {"error": ( + f"Plugin standalone send for '{platform_name}' returned an " + f"invalid result: expected a dict with 'success' or 'error' " + f"keys, got {type(result).__name__}")} async def _send_chunks(chunks, send_one): @@ -651,10 +506,10 @@ async def _send_chunks(chunks, send_one): def _platform_max_length(platform): - """Max message length for chunking: the adapter constant for Signal (its - raw JSON-RPC path never sees SignalAdapter's own chunking, so signal-cli - would reject long sends whole), the registry's ``max_message_length`` for - plugins (Slack, Feishu, ...), else None (no chunking).""" + """Max message length for chunking: the adapter constant for Signal (its raw + JSON-RPC path never sees SignalAdapter's own chunking, so signal-cli would reject + long sends whole), the registry's ``max_message_length`` for plugins (Slack, + Feishu, ...), else None (no chunking).""" from gateway.config import Platform if platform == Platform.SIGNAL: @@ -673,20 +528,18 @@ def _platform_max_length(platform): return None -# Plugin platforms whose media (and, for Discord, all) sends go straight to the -# registry ``standalone_sender_fn`` — bypassing the live adapter on purpose: -# Discord's plugin ``_standalone_send`` handles forum channels, threads and -# multipart uploads (historically the only Discord path); Slack uploads via -# files_upload_v2; WhatsApp posts each file to the Baileys bridge /send-media so -# images/videos/audio arrive as native bubbles, not documents. +# Plugin platforms whose media (and, for Discord, all) sends go straight to the registry +# ``standalone_sender_fn`` — bypassing the live adapter on purpose: Discord's plugin +# ``_standalone_send`` handles forum channels, threads and multipart uploads +# (historically the only Discord path); Slack uploads via files_upload_v2; WhatsApp +# posts each file to the Baileys bridge /send-media so media arrive as native bubbles. # platform -> (error label, run discover_plugins first, caption-capable, # media_files sentinel for non-final chunks, forward force_document) _PLUGIN_STANDALONE_MEDIA = { "discord": ("Discord", False, True, [], False), "feishu": ("Feishu", True, False, None, False), "slack": ("Slack", True, True, [], False), - "whatsapp": ("WhatsApp", True, True, None, True), -} + "whatsapp": ("WhatsApp", True, True, None, True)} async def _send_plugin_standalone( @@ -702,36 +555,35 @@ async def _send_plugin_standalone( if captionable: # Cap on the platform's own message limit so the caption is deliverable. caption, _ = _media_caption_split( - message, media_files, max_caption_len=(max_len or _DEFAULT_CAPTION_LIMIT) - ) + message, media_files, max_caption_len=(max_len or _DEFAULT_CAPTION_LIMIT)) if caption is not None: return await sender( pconfig, chat_id, "", thread_id=thread_id, media_files=media_files, - caption=caption, **extra, - ) + caption=caption, **extra) return await _send_chunks( chunks, lambda chunk, is_last: sender( pconfig, chat_id, chunk, thread_id=thread_id, - media_files=media_files if is_last else empty_media, **extra, - ), - ) + media_files=media_files if is_last else empty_media, **extra)) -# Native-media chunked routes for built-in platforms; media rides on the final -# chunk, non-final chunks get the platform's empty-media sentinel. +# Native-media chunked routes for built-in platforms; media rides on the final chunk, +# non-final chunks get the platform's empty-media sentinel. # platform -> (media required, empty-media sentinel, # sender(platform, pconfig, chat_id, chunk, media, thread_id, force_document)) -# Matrix: ALL sends go through the native adapter so text is encrypted in E2EE -# rooms too (the raw-HTTP standalone path is not encryption-aware). +# Matrix: ALL sends go through the native adapter so text is encrypted in E2EE rooms +# too (the raw-HTTP standalone path is not encryption-aware). # Signal: attachments ride the JSON-RPC ``attachments`` param. # Yuanbao: media needs the running gateway adapter's WebSocket. -# Slack (text; media was intercepted above): prefer the live adapter — it is -# multi-workspace aware and honors adapter gates like ignored_channels, while -# the standalone Web-API path may only have a token list — else the plugin's -# standalone sender, keeping MEDIA delivery on the cron fallback. +# Slack (text; media was intercepted above): prefer the live adapter — multi-workspace +# aware and honors adapter gates like ignored_channels, while the standalone Web-API +# path may only have a token list — else the plugin's standalone sender. # WeCom: native media only through the live gateway adapter. # Names resolve at call time so tests can monkeypatch e.g. ``_send_signal``. +def _via_adapter_route(p, pc, cid, chunk, media, tid, fd): + return _send_via_adapter(p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd) + + _CHUNKED_ROUTES = { "matrix": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_matrix_via_adapter( pc, cid, chunk, media_files=media, thread_id=tid)), @@ -739,26 +591,22 @@ _CHUNKED_ROUTES = { pc.extra, cid, chunk, media_files=media)), "yuanbao": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_yuanbao( cid, chunk, media_files=media)), - "slack": (False, [], lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter( - p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)), - "wecom": (True, None, lambda p, pc, cid, chunk, media, tid, fd: _send_via_adapter( - p, pc, cid, chunk, thread_id=tid, media_files=media, force_document=fd)), -} + "slack": (False, [], _via_adapter_route), + "wecom": (True, None, _via_adapter_route)} -# Text-only senders for built-in platforms (generic path; media is dropped -# with a warning). Signature: (pconfig, chat_id, chunk, thread_id) -> result. + +def _registry_text_sender(name): + return lambda pc, cid, chunk, tid: _registry_standalone_send(name, pc, cid, chunk, tid) + + +# Text-only senders for built-in platforms (generic path; media is dropped with a +# warning). Signature: (pconfig, chat_id, chunk, thread_id) -> result. _TEXT_SENDERS = { - "whatsapp": lambda pc, cid, chunk, tid: _registry_standalone_send("whatsapp", pc, cid, chunk, tid), + **{name: _registry_text_sender(name) for name in ("whatsapp", "email", "sms", "dingtalk", "feishu", "wecom")}, "signal": lambda pc, cid, chunk, tid: _send_signal(pc.extra, cid, chunk), - "email": lambda pc, cid, chunk, tid: _registry_standalone_send("email", pc, cid, chunk, tid), - "sms": lambda pc, cid, chunk, tid: _registry_standalone_send("sms", pc, cid, chunk, tid), - "dingtalk": lambda pc, cid, chunk, tid: _registry_standalone_send("dingtalk", pc, cid, chunk, tid), - "feishu": lambda pc, cid, chunk, tid: _registry_standalone_send("feishu", pc, cid, chunk, tid), - "wecom": lambda pc, cid, chunk, tid: _registry_standalone_send("wecom", pc, cid, chunk, tid), "bluebubbles": lambda pc, cid, chunk, tid: _send_bluebubbles(pc.extra, cid, chunk), "qqbot": lambda pc, cid, chunk, tid: _send_qqbot(pc, cid, chunk), - "yuanbao": lambda pc, cid, chunk, tid: _send_yuanbao(cid, chunk), -} + "yuanbao": lambda pc, cid, chunk, tid: _send_yuanbao(cid, chunk)} _MEDIA_PLATFORMS_NOTE = "telegram, discord, matrix, weixin, signal, yuanbao, feishu, whatsapp and slack" @@ -766,12 +614,11 @@ _MEDIA_PLATFORMS_NOTE = "telegram, discord, matrix, weixin, signal, yuanbao, fei async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False, args=None): """Route a message to the appropriate platform sender. - Long messages are chunked with the adapters' smart splitter (code-block - aware, part indicators). Branch order matters: Weixin first (its native - helper must not be blocked by unrelated optional imports such as - lark-oapi's heavy Feishu path), Telegram (chunks itself), plugin - standalone media routes, native chunked routes, then the generic text - path that drops media with a warning. + Long messages are chunked with the adapters' smart splitter (code-block aware, + part indicators). Branch order matters: Weixin first (its native helper must not + be blocked by unrelated optional imports such as lark-oapi's heavy Feishu path), + Telegram (chunks itself), plugin standalone media routes, native chunked routes, + then the generic text path that drops media with a warning. """ from gateway.config import Platform @@ -781,20 +628,14 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if platform == Platform.WEIXIN: return await _send_weixin(pconfig, chat_id, message, media_files=media_files) - # Telegram chunks internally on the *formatted* text in UTF-16 units - # (MarkdownV2/HTML escaping inflates length), so it gets the whole - # message; media attaches after all text chunks. + # Telegram chunks internally on the *formatted* text in UTF-16 units (MarkdownV2/HTML + # escaping inflates length), so it gets the whole message; media attaches after all text. if platform == Platform.TELEGRAM: disable_link_previews = bool(getattr(pconfig, "extra", {}) and pconfig.extra.get("disable_link_previews")) return await _send_telegram( - pconfig.token, - chat_id, - message, - media_files=media_files, - thread_id=thread_id, - disable_link_previews=disable_link_previews, - force_document=force_document, - ) + pconfig.token, chat_id, message, + media_files=media_files, thread_id=thread_id, + disable_link_previews=disable_link_previews, force_document=force_document) from gateway.platforms.base import BasePlatformAdapter @@ -804,33 +645,27 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, if platform_name == "discord" or (media_files and platform_name in _PLUGIN_STANDALONE_MEDIA): return await _send_plugin_standalone( platform_name, pconfig, chat_id, message, chunks, media_files, - thread_id=thread_id, max_len=max_len, force_document=force_document, - ) + thread_id=thread_id, max_len=max_len, force_document=force_document) route = _CHUNKED_ROUTES.get(platform_name) if route is not None and (media_files or not route[0]): _, empty_media, sender = route return await _send_chunks(chunks, lambda chunk, is_last: sender( platform, pconfig, chat_id, chunk, - media_files if is_last else empty_media, thread_id, force_document, - )) + media_files if is_last else empty_media, thread_id, force_document)) - # --- Generic path: text only. Buzz is a plugin platform with verified - # native media delivery through _send_via_adapter (media-only sends - # included), so it is exempt from the media error/warning. - if media_files and platform_name != "buzz" and not message.strip(): - return { - "error": ( - f"send_message MEDIA delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}; " - f"target {platform_name} had only media attachments" - ) - } + # --- Generic path: text only. Buzz is a plugin platform with verified native media + # delivery through _send_via_adapter (media-only sends included), so it is exempt + # from the media error/warning. warning = None if media_files and platform_name != "buzz": + if not message.strip(): + return {"error": ( + f"send_message MEDIA delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}; " + f"target {platform_name} had only media attachments")} warning = ( f"MEDIA attachments were omitted for {platform_name}; " - f"native send_message media delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}" - ) + f"native send_message media delivery is currently only supported for {_MEDIA_PLATFORMS_NOTE}") text_sender = _TEXT_SENDERS.get(platform_name) if text_sender is not None: @@ -852,27 +687,11 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, except Exception as e: return {"error": f"Plugin send_message handler failed: {e}"} # Plugin platform: live gateway adapter if available, else standalone_sender_fn. - send_one = lambda chunk, is_last: _send_via_adapter( # noqa: E731 - platform, pconfig, chat_id, chunk, thread_id=thread_id, - media_files=media_files if is_last else [], force_document=force_document, - ) + send_one = lambda chunk, is_last: _via_adapter_route( # noqa: E731 + platform, pconfig, chat_id, chunk, media_files if is_last else [], thread_id, force_document) last_result = await _send_chunks(chunks, send_one) - if ( - warning - and isinstance(last_result, dict) - and last_result.get("success") - and not last_result.get("media_delivered") - ): + if (warning and isinstance(last_result, dict) and last_result.get("success") + and not last_result.get("media_delivered")): last_result["warnings"] = [*last_result.get("warnings", []), warning] return last_result - - -# --- Registry --- -from tools.registry import tool_error - -# NOTE: ``send_message`` is intentionally NOT registered as an agent-callable -# model tool (the agent must not fire cross-platform messages on its own). -# The send engine here is the shared transport for cron delivery, the -# ``hermes send`` CLI, the gateway kanban notifier and the opt-in MCP server, -# which import the helpers directly.