"""Send Message Tool -- cross-channel messaging via platform APIs (send, list targets, react); works in both CLI and gateway contexts.""" import asyncio import json import logging import os from functools import partial from agent.secret_scope import get_secret logger = logging.getLogger(__name__) from tools.send_message_targets import _HOME_CHANNEL_ENV_OVERRIDES, _SLACK_USER_ID_RE, resolve_send_target from tools.send_message_senders import ( _AUDIO_EXTS, _DEFAULT_CAPTION_LIMIT, _IMAGE_EXTS, _NO_DELIVERABLE, _VIDEO_EXTS, _VOICE_EXTS, _adapter_media_method, _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); cron delivery, the # ``hermes send`` CLI, the kanban notifier and the opt-in MCP server import the helpers. def prepare_send_message_platforms() -> None: """Load enabled standalone plugins before tool schemas/cache keys are built.""" from hermes_cli.plugins import discover_plugins discover_plugins() def send_message_tool(args, **kw): """Handle cross-channel send_message tool calls.""" action = args.get("action", "send") if action == "list": return _handle_list() if action in ("react", "unreact"): return _handle_react(args, remove=action == "unreact") return _handle_send(args) def _resolve_tool_target(target: str, *, pass_unresolved_references: bool = False): """``(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 = target.partition(":") platform_name, target_ref = platform_name.strip().lower(), target_ref.strip() or None prepare_send_message_platforms() if not target_ref: return platform_name, None, None, None return platform_name, *resolve_send_target(platform_name, target_ref, pass_unresolved_references=pass_unresolved_references) def _handle_list(): try: from gateway.channel_directory import format_directory_for_display return json.dumps({"targets": format_directory_for_display()}) except Exception as e: return json.dumps(_error(f"Failed to load channel directory: {e}")) _TOKEN_UNSET = object() def _authorize_relay_target(platform_name: str, chat_id, thread_id=None, *, native_token=_TOKEN_UNSET) -> str | None: """Relay egress-authorization guard (P5a); None when the send may proceed. Thin delegate to ``gateway.relay.egress`` so the tool keeps working in environments where the gateway package can't be imported. THE TWO FAILURES ARE NOT THE SAME, and conflating them disabled the boundary. A missing gateway module means there is no relay egress to authorize, so proceeding is correct. A fault INSIDE the guard means authorization did not happen — and returning None there means "authorized", so a single runtime bug in the guard silently switched the whole P5(a) boundary off. Review found this by making the guard raise and watching the send go through. So: the import is tolerated, the CALL is not. A guard that cannot answer refuses, which is the only safe polarity for an authorization check. """ try: from gateway.relay.egress import authorize_relay_target except ImportError as exc: # ABSENCE ONLY, and absence means the gateway relay module ITSELF is # missing — `exc.name` says which module was not found. An ImportError # naming a NESTED dependency is a broken installation, i.e. a fault, # and returning None here means "authorized". Review probed exactly # that (`ImportError.name = "gateway.relay.dependency"`) and got an # authorized verdict, so `except ImportError` alone was still fail-open. # ABSENCE has one shape and it is checkable: a genuinely missing module # raises ModuleNotFoundError with `.name` set to the module that was not # found (verified: `import gateway.relay.x` -> ModuleNotFoundError, # name="gateway.relay.x"). So a plain ImportError, or a nameless one, is # an unattributable FAULT — never proof that there is no relay here. # I previously admitted the nameless case to protect the CLI/cron path; # that reasoning was wrong, because that path does not produce one. _missing = getattr(exc, "name", None) if not isinstance(exc, ModuleNotFoundError) or _missing not in ( "gateway", "gateway.relay", "gateway.relay.egress", ): logger.exception( "relay egress module failed to import for %s — refusing the send", platform_name, ) return ( f"Refusing to send to relay target '{platform_name}': the egress " "authorization module could not be loaded, so this destination " "could not be verified." ) logger.debug("relay target authorization unavailable", exc_info=True) return None except Exception: # noqa: BLE001 - the module is THERE and broke; FAIL CLOSED logger.exception( "relay egress module failed to import for %s — refusing the send", platform_name, ) return ( f"Refusing to send to relay target '{platform_name}': the egress " "authorization module could not be loaded, so this destination " "could not be verified." ) try: # ONE SNAPSHOT. `native_token` is the token from the SAME pconfig the # dispatch below will actually send with. Letting the guard reload # config independently allowed a transition where authorization saw a # connector-only setup (exemption granted) while dispatch still held a # native token and sent the unattested handle itself. if native_token is _TOKEN_UNSET: # A caller that forgets the snapshot must NOT silently look like # "no native token", which would grant the @handle exemption. return authorize_relay_target(platform_name, chat_id, thread_id) return authorize_relay_target( platform_name, chat_id, thread_id, native_token=native_token ) except Exception: # noqa: BLE001 - the guard faulted; FAIL CLOSED logger.exception( "relay target authorization FAILED for %s — refusing the send", platform_name, ) return ( f"Refusing to send to relay target '{platform_name}': the egress " "authorization check failed, so this destination could not be " "verified. This is a bug — the send was blocked rather than " "allowed through unchecked." ) def _handle_react(args, remove=False): """Attach (``remove=True``: retract) an emoji reaction via the live gateway adapter; no standalone fallback because reacting needs the adapter's live message-id state.""" target, emoji = args.get("target", ""), (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("'target' is required when action='unreact'" if remove else "Both 'target' and 'emoji' are required when action='react'") # Platform-native ids (e.g. photon GUIDs) match no parser/directory entry; the adapter validates. platform_name, chat_id, _thread_id, resolution_error = _resolve_tool_target(target, pass_unresolved_references=True) if resolution_error: return tool_error(resolution_error) platform, err = _platform_enum(platform_name) if err: return tool_error(err) if not chat_id: try: from gateway.config import load_gateway_config chat_id = load_gateway_config().get_home_channel(platform).chat_id except Exception: return tool_error(f"No chat specified and no home channel set for {platform_name}. " f"Use '{platform_name}:chat_id'.") # P5(a): same egress-authorization floor as the send path — a reaction is # an outbound act against a named destination, so an unattested relay # target must be refused here too, not just on `send`. # The react path has no pconfig snapshot of its own; it dispatches through # the LIVE adapter below, never through a native token, so the guard does # its own credential probe here. _relay_denial = _authorize_relay_target(platform_name, chat_id, _thread_id) if _relay_denial: return tool_error(_relay_denial) _, 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).") 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.") try: from model_tools import _run_async result = _run_async(react_fn(chat_id=chat_id, message_id=message_id, **({} if remove else {"emoji": emoji}))) except Exception as e: return json.dumps(_error(f"Reaction failed: {e}")) return json.dumps(result if isinstance(result, dict) else {"success": bool(result)}) def _handle_send(args): target, message = args.get("target", ""), args.get("message", "") if not target or not message: return tool_error("Both 'target' and 'message' are required when action='send'") # Lone surrogates reach the outbound body via surrogateescape-decoded argv # (`hermes send` MESSAGE) and crash the UTF-8 marshal inside platform SDK # request bodies (feishu/lark, #113799). Every send_message caller (model tool # call, `hermes send`, dashboard console) enters here, so scrub once before the # media extraction, the session mirror and the platform sender see the text. # Model output delivered by the gateway/cron is already scrubbed upstream # (``agent/turn_finalizer.py::finalize_turn``, ``gateway/run.py``). from agent.message_sanitization import _sanitize_surrogates message = _sanitize_surrogates(message) 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(): return tool_error("Interrupted") try: from gateway.config import load_gateway_config config = load_gateway_config() except Exception as e: return json.dumps(_error(f"Failed to load gateway config: {e}")) platform, pconfig, entry, err = _resolve_platform_config(platform_name, config) if err: return tool_error(err) from gateway.platforms.base import BasePlatformAdapter # Capture [[as_document]] before extract_media strips it (images keep original bytes via send_document). force_document_attachments = "[[as_document]]" in message media_files, cleaned_message = BasePlatformAdapter.extract_media(message) media_dropped: list = [] media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files, dropped=media_dropped) mirror_text = cleaned_message.strip() or _describe_media_for_mirror(media_files) used_home_channel = not chat_id if used_home_channel: chat_id, err = _home_chat_id(config, platform, platform_name) if err: return tool_error(err) if duplicate_skip := _maybe_skip_cron_duplicate_send(platform_name, chat_id, thread_id): return json.dumps(duplicate_skip) # Slack: resolve user targets to DM channel IDs before sending. _parse_target_ref emits internal # ``user:U...`` / ``user_name:@handle`` targets; a bare U... id can also arrive from session metadata or # the home-channel config. All are opened via conversations.open (fixes #19236). if platform_name == "slack" and chat_id: chat_id, resolve_err = _slack_dm_chat_id(pconfig, chat_id) if resolve_err: return json.dumps(resolve_err) # POSITION IS LOAD-BEARING — this must stay BELOW Slack user→DM resolution. # `_parse_target_ref` emits internal pseudo-ids (`user_name:ben`, # `user:U...`) that no provenance can ever contain, because provenances # record RESOLVED conversation ids. Authorizing above the resolver compared # a handle against a set of `D...` ids and refused every Slack DM — a fix # that caused the outage it was meant to prevent. Pinned by # test_slack_user_targets_resolve_then_authorize; moving this call back up # turns those cases red. # thread_id is part of the DESTINATION: on Discord the thread is the literal # REST target, so an attested parent must not vouch for an arbitrary thread. _relay_denial = _authorize_relay_target(platform_name, chat_id, thread_id, native_token=getattr(pconfig, "token", None)) if _relay_denial: return tool_error(_relay_denial) try: from model_tools import _run_async # Only custom plugin handlers receive the complete typed request. ``mentions`` is a WhatsApp-only # contract (the CLI rejects it elsewhere); other platforms' standalone senders don't accept the kwarg. handler_args = {"args": args} if entry is not None and entry.send_message_handler is not None else {} mentions = args.get("mentions") if mentions and platform_name == "whatsapp": handler_args["mentions"] = [mentions] if isinstance(mentions, str) else list(mentions) result = _run_async(_send_to_platform(platform, pconfig, chat_id, cleaned_message, thread_id=thread_id, media_files=media_files, force_document=force_document_attachments, **handler_args)) 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})" if mirror_text and _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id): result["mirrored"] = True if media_dropped: # The text went out but an attachment the caller asked for did not: a script reading # ``success`` / exit 0 must not book a delivery that never happened (#115908). result["success"] = False result["partial_success"] = True result["error"] = (f"Delivery incomplete: {len(media_dropped)} requested MEDIA attachment(s) " "dropped before delivery (see media_dropped)") if isinstance(result, dict) and media_dropped: result["media_dropped"] = media_dropped if isinstance(result, dict) and "error" in result: result["error"] = _sanitize_error_text(result["error"]) return json.dumps(result) except Exception as e: return json.dumps(_error(f"Send failed: {e}")) def _platform_enum(platform_name): """``(Platform, None)`` or ``(None, error)`` for a platform name.""" from gateway.config import Platform try: return Platform(platform_name), None except (ValueError, KeyError): return None, f"Unknown platform: {platform_name}" def _resolve_platform_config(platform_name, config): """``(platform, pconfig, registry_entry, error)``. Plugin platforms must be registered; disabled/missing platforms error, except Weixin, which may be configured purely via .env.""" from gateway.config import Platform from gateway.platform_registry import platform_registry entry = platform_registry.get(platform_name) if entry is None and platform_name not in {member.value for member in Platform}: return None, None, None, f"Unknown or unregistered plugin platform: {platform_name}" platform, err = _platform_enum(platform_name) if err: return None, None, None, err pconfig = config.platforms.get(platform) if not pconfig or not pconfig.enabled: pconfig = _weixin_env_pconfig() if platform_name == "weixin" else None if pconfig is None: return None, None, None, _not_configured_error(platform_name, platform, entry) return platform, pconfig, entry, None def _not_configured_error(platform_name, platform, entry): """Name the resolved home and what each credential source held, so the user edits the file this process actually read (a hardcoded ``~/.hermes`` does not exist on a Windows or profile home).""" from agent.secret_scope import load_env_file from gateway.config import _getenv from gateway.config_env import _ENV_ENABLE_CREDENTIALS from hermes_constants import get_hermes_home home = get_hermes_home() env_names = list(_ENV_ENABLE_CREDENTIALS.get(platform) or (entry.required_env if entry else ())) names = "/".join(env_names) or "credentials" env_path, config_path = home / ".env", home / "config.yaml" dotenv_keys = load_env_file(env_path) dotenv_state = (f"{names} present" if any(n in dotenv_keys for n in env_names) else f"no {names}") \ if env_path.exists() else "missing" try: from hermes_cli.config_effective import load_user_config_effective user_config = load_user_config_effective(config_path) or {} block = user_config.get("platforms", {}).get(platform_name) except Exception: user_config, block = {}, None if not config_path.exists(): config_state = "missing" elif not isinstance(block, dict): config_state = f"no platforms.{platform_name} block" elif block.get("enabled") is False: config_state = f"platforms.{platform_name}.enabled: false" else: config_state = f"platforms.{platform_name} has no token" env_state = f"{names} set" if any(_getenv(n) for n in env_names) else f"{names} unset" msg = (f"Platform '{platform_name}' is not configured. Looked in: {env_path} ({dotenv_state}), " f"{config_path} ({config_state}), environment ({env_state}), " f"external secret sources ({_secret_sources_state(user_config)}).") # The gateway can hold a token only in its own process environment; a fresh CLI cannot see it. A # gateway started from the default root (the reporter's shell had HERMES_HOME=/profiles/

) # never reads this profile's .env at all. try: from gateway.status import read_runtime_status, runtime_status_pid_is_live from hermes_constants import get_default_hermes_root, hermes_home_key root = get_default_hermes_root() gateways = [(home, read_runtime_status())] if hermes_home_key(root) != hermes_home_key(home): gateways.append((root, read_runtime_status(root / "gateway_state.json"))) for gw_home, record in gateways: state = ((record or {}).get("platforms") or {}).get(platform_name, {}).get("state") if state != "connected" or "present" in dotenv_state or not runtime_status_pid_is_live(record): continue msg += f" A gateway (pid {record.get('pid')}) running from {gw_home} has {platform_name} connected" msg += (f", so its credentials live only in that process's environment; add {names} to {env_path}." if gw_home is home else f"; this shell is scoped to profile home {home} whose .env has no {names}.") except Exception: pass return msg def _secret_sources_state(user_config): """``name: enabled|disabled`` for every registered secret source with a ``secrets.`` section, or ``none configured``; names only, never values.""" try: from agent.secret_sources.registry import list_sources secrets_cfg = user_config.get("secrets") if isinstance(user_config.get("secrets"), dict) else {} states = [f"{s.name}: {'enabled' if s.is_enabled(secrets_cfg[s.name]) else 'disabled'}" for s in list_sources() if isinstance(secrets_cfg.get(s.name), dict)] except Exception: states = [] return ", ".join(states) or "none configured" def _home_chat_id(config, platform, platform_name): """``(home chat_id, None)`` or ``(None, actionable error)``; Weixin also honours WEIXIN_HOME_CHANNEL.""" home = config.get_home_channel(platform) if home: return home.chat_id, None # Home channel is a per-profile target like the token beside it: a raw environ read would post a # multiplexed secondary's message into the default profile's Weixin chat. wx_home = (get_secret("WEIXIN_HOME_CHANNEL", "") or "").strip() if platform_name == "weixin" else "" if wx_home: return wx_home, None 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} ") def _slack_dm_chat_id(pconfig, chat_id): """Open Slack user targets (``user:``/``user_name:`` from the parser, or a bare U... id from session metadata / home-channel config) as DM conversations. ``(chat_id, None)`` or ``(None, error_dict)``.""" dm_target = f"user:{chat_id}" if chat_id.startswith("U") and _SLACK_USER_ID_RE.fullmatch(chat_id) else chat_id if not dm_target.startswith(("user:", "user_name:")): return chat_id, None from model_tools import _run_async return _run_async(_resolve_slack_user_target(pconfig.token, dm_target)) def _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id): """Best-effort mirror of the sent message into the target's gateway session.""" try: 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, thread_id=thread_id, source_label=get_session_env("HERMES_SESSION_PLATFORM", "cli"), user_id=get_session_env("HERMES_SESSION_USER_ID", "") or None)) except Exception: return False def _weixin_env_pconfig(): """Synthesize a Weixin PlatformConfig from .env secrets, or None.""" wx_token = get_secret("WEIXIN_TOKEN", "").strip() wx_account = get_secret("WEIXIN_ACCOUNT_ID", "").strip() 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()}) def _describe_media_for_mirror(media_files): """Return a human-readable mirror summary when a message only contains media.""" if not media_files: return "" if len(media_files) != 1: return f"[Sent {len(media_files)} media attachments]" media_path, is_voice = media_files[0] ext = os.path.splitext(media_path)[1].lower() if is_voice and ext in _VOICE_EXTS: return "[Sent voice message]" kind = next((k for exts, k in ((_IMAGE_EXTS, "image"), (_VIDEO_EXTS, "video"), (_AUDIO_EXTS, "audio")) if ext in exts), "document") return f"[Sent {kind} attachment]" 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.""" 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 and auto_chat_id and auto_platform == platform_name and auto_chat_id == str(chat_id) and (get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID", "").strip() or None) == thread_id): return None target_label = f"{platform_name}:{chat_id}" + (f":{thread_id}" if thread_id is not None else "") return {"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") return text if len(text) <= max_chars else f"{text[: max_chars - 3]}..." async def _send_live_adapter_media(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 kind are unsupported, not no-op'd.""" caption, separate_text = _media_caption_split(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) 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): media_path = descriptor[0] if isinstance(descriptor, (list, tuple)) and descriptor else None if not isinstance(media_path, str) or not media_path: return {"error": f"Adapter media send failed: invalid media descriptor {index + 1}/{total}"} is_voice = len(descriptor) > 1 and bool(descriptor[1]) 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() method_name, media_kind = _adapter_media_method(ext, is_voice or ext in _AUDIO_EXTS, force_document) adapter_method = getattr(type(adapter), method_name, None) 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")} try: last_result = await getattr(adapter, method_name)( chat_id, media_path, caption=caption if index == 0 else None, reply_to=thread_id, metadata=metadata) except asyncio.CancelledError: raise except Exception as exc: 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}"} if last_result is None: return {"error": _NO_DELIVERABLE} return {"success": True, "message_id": last_result.message_id, "media_delivered": True} async def _dispatch_on_gateway_loop(runner, make_coro, log_message): """Await ``make_coro()`` on the gateway's loop: adapter.send() uses queues/tasks bound to it, so awaiting from another loop (the tool worker thread) deadlocks.""" gateway_loop = getattr(runner, "_gateway_loop", None) if gateway_loop is None or asyncio.get_running_loop() is gateway_loop: return await make_coro() # same loop / no gateway loop (CLI, tests) if not gateway_loop.is_running(): 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) if fut is None: return {"error": "Gateway loop unavailable for send dispatch"} # shield: a cancelled caller must not cancel the enqueued send (a retry would duplicate it). # No timeout: the adapter and outer _run_async bound the wait. return await asyncio.shield(asyncio.wrap_future(fut)) async def _send_via_adapter(platform, pconfig, chat_id, chunk, *, thread_id=None, media_files=None, force_document=False): """Live in-process gateway adapter first, else the plugin's ``standalone_sender_fn`` (cron), else an error naming both; media uses the adapter's native media APIs under the same rules.""" platform_name = platform.value if hasattr(platform, "value") else str(platform) runner, adapter = _live_adapter(platform) if adapter is not None: try: metadata = {**({"thread_id": thread_id} if thread_id else {}), **({"publish_topic": chat_id} if platform_name == "ntfy" and chat_id else {})} or None if media_files: # always a dict result, returned as-is below make_coro = lambda: _send_live_adapter_media( # noqa: E731 adapter, chat_id, chunk, media_files, thread_id=thread_id, metadata=metadata, force_document=force_document) else: make_coro = lambda: adapter.send(chat_id=chat_id, content=chunk, metadata=metadata) # noqa: E731 result = await _dispatch_on_gateway_loop( runner, make_coro, f"send_message: failed to schedule{' media send' if media_files else ''} on gateway loop") except asyncio.CancelledError: raise except Exception as e: return {"error": f"Plugin platform send failed: {_bounded_send_error(e)}"} if isinstance(result, dict): return result if result.success: return {"success": True, "message_id": result.message_id} return {"error": f"Adapter send failed: {_bounded_send_error(result.error)}"} try: from gateway.platform_registry import platform_registry sender = platform_registry.get(platform_name).standalone_sender_fn except Exception: sender = None if sender is None: return {"error": (f"No live adapter for platform '{platform_name}'. Is the gateway running with this platform " f"connected? For out-of-process delivery (e.g. cron in a separate process), the platform " f"plugin must register a standalone_sender_fn on its PlatformEntry.")} try: result = await sender(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")): return {**result, "error": _bounded_send_error(result["error"])} if result.get("error") else result return {"error": (f"Plugin standalone send for '{platform_name}' returned an invalid result: " f"expected a dict with 'success' or 'error' keys, got {type(result).__name__}")} async def _send_chunks(chunks, send_one): """``send_one(chunk, is_last)`` in order; stop at the first error dict, else last result.""" result = None # --- Matrix: route ALL sends through the native adapter so text is encrypted in E2EE rooms too (issue: # text-only sends arrived with a red padlock because they took the raw-HTTP standalone path). The # adapter reuses the live gateway's E2EE session when available (#46310) and falls back to an # encryption-aware ephemeral adapter for standalone/cron. --- for i, chunk in enumerate(chunks): result = await send_one(chunk, i == len(chunks) - 1) if isinstance(result, dict) and result.get("error"): break return result def _platform_max_length(platform): """Chunking limit: Signal's adapter constant (its raw JSON-RPC path bypasses the adapter's chunking), the registry's ``max_message_length`` for plugins, else None (no chunking).""" from gateway.config import Platform if platform == Platform.SIGNAL: try: from gateway.platforms.signal import MAX_MESSAGE_LENGTH return MAX_MESSAGE_LENGTH except ImportError: return 8000 try: from gateway.platform_registry import platform_registry entry = platform_registry.get(platform.value) return entry.max_message_length if entry and entry.max_message_length > 0 else None except Exception: return None # Plugin platforms whose media (Discord: all) sends deliberately bypass the live adapter for the # registry ``standalone_sender_fn`` (Discord: forums/threads/multipart; Slack: files_upload_v2; # WhatsApp: Baileys /send-media). 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)} async def _send_plugin_standalone(platform_name, pconfig, chat_id, message, chunks, media_files, *, thread_id, max_len, force_document, mentions=None): """Chunked send through a plugin's standalone_sender_fn; one captionable file + short text rides as the media caption. WhatsApp re-pings recipients on every message that carries ``mentions``, so only the first payload of a logical send gets them.""" label, discover, captionable, empty_media, pass_force = _PLUGIN_STANDALONE_MEDIA[platform_name] sender, err = _plugin_standalone_sender(platform_name, label=label, discover=discover) if err: return err extra = {"force_document": force_document} if pass_force else {} first_only = {"mentions": mentions} if mentions else {} 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)) if caption is not None: return await sender(pconfig, chat_id, "", thread_id=thread_id, media_files=media_files, caption=caption, **extra, **first_only) def send_one(chunk, is_last): kwargs = {**extra, **first_only} first_only.clear() return sender(pconfig, chat_id, chunk, thread_id=thread_id, media_files=media_files if is_last else empty_media, **kwargs) return await _send_chunks(chunks, send_one) 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) # Native-media chunked routes for built-in platforms; media rides on the final chunk, non-final # chunks get the sentinel. platform -> (media required, sentinel, sender(platform, pconfig, # chat_id, chunk, media, thread_id, force_document)). Matrix: ALL sends use the native adapter # (E2EE text). Signal: attachments ride the JSON-RPC param. Yuanbao / WeCom: media needs the # running gateway. Slack text: live adapter (multi-workspace, ignored_channels gates) else the # plugin's standalone sender. Names resolve at call time so tests can monkeypatch ``_send_signal``. _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)), "signal": (True, [], lambda p, pc, cid, chunk, media, tid, fd: _send_signal( 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, [], _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. _TEXT_SENDERS = { **{name: partial(_registry_standalone_send, name) for name in ("whatsapp", "email", "sms", "dingtalk", "feishu", "wecom")}, "signal": lambda pc, cid, chunk, tid: _send_signal(pc.extra, cid, chunk), "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)} _MEDIA_PLATFORMS_NOTE = "telegram, discord, matrix, weixin, signal, yuanbao, feishu, whatsapp and slack" async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False, mentions=None, args=None): """Route to the platform sender, chunking long text with the adapters' splitter. Order matters: Weixin first (its native helper must not be blocked by unrelated optional imports such as lark-oapi), Telegram (chunks itself), plugin standalone media, native chunked, generic text.""" from gateway.config import Platform platform_name = platform.value if hasattr(platform, "value") else str(platform) media_files = media_files or [] if platform == Platform.WEIXIN: return await _send_weixin(pconfig, chat_id, message, media_files=media_files) # Telegram chunks internally on the *formatted* text (escaping inflates length). if platform == Platform.TELEGRAM: return await _send_telegram( pconfig.token, chat_id, message, media_files=media_files, thread_id=thread_id, force_document=force_document, disable_link_previews=bool(getattr(pconfig, "extra", {}) and pconfig.extra.get("disable_link_previews"))) from gateway.platforms.base import BasePlatformAdapter max_len = _platform_max_length(platform) chunks = BasePlatformAdapter.truncate_message(message, max_len) if max_len else [message] if (platform_name == "discord" or (platform_name == "whatsapp" and mentions) 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, mentions=mentions) 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)) # Generic path: text only. Buzz delivers media natively via _send_via_adapter, so no 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}") text_sender = _TEXT_SENDERS.get(platform_name) if text_sender is not None: send_one = lambda chunk, is_last: text_sender(pconfig, chat_id, chunk, thread_id) # noqa: E731 else: from gateway.platform_registry import platform_registry entry = platform_registry.get(platform_name) if entry is not None and entry.send_message_handler is not None: # Custom handler receives the full typed request once (not per chunk). try: import inspect result = entry.send_message_handler(args or {}, chat_id, platform_name, pconfig) return await result if inspect.isawaitable(result) else result 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: _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")): last_result["warnings"] = [*last_result.get("warnings", []), warning] return last_result