"""WhatsApp platform adapter (Baileys bridge): a Node.js bridge process runs the WhatsApp Web client; messages are polled over a local HTTP API and responses are posted back through it.""" import asyncio import logging import mimetypes import os import platform import re import signal import subprocess from contextlib import suppress from functools import wraps from pathlib import Path from typing import Dict, Optional, Any from gateway.platforms._shared import ( apply_yaml_bridge as _apply_yaml_bridge, extra_or_secret as _extra_or_secret, get_scoped_secret, send_error ) from hermes_cli._subprocess_compat import windows_detach_popen_kwargs from hermes_constants import (find_node_executable, get_hermes_dir, with_hermes_node_path) _IS_WINDOWS = platform.system() == "Windows" def _wenv(name: str, default: str = "") -> str: """WHATSAPP_* env via the profile secret scope (multiplexed profiles see their own .env, not the process-global value).""" return get_scoped_secret(name, default) logger = logging.getLogger(__name__) # Owner-typed inbound text is prefixed at MessageEvent construction so transcripts stay disambiguated before silent_ingest. _OWNER_REPLY_PREFIX = "[owner reply] " _RUN_TEXT = dict(capture_output=True, text=True, encoding='utf-8', errors='replace', stdin=subprocess.DEVNULL) def _listener_pids_on_port(port: int) -> list: """PIDs *listening* on ``port`` (POSIX), never clients — a bare ``lsof -i :PORT`` once killed the user's browser.""" pids: list = [] with suppress(FileNotFoundError): # lsof not installed — fall through to ss pids = _safe_ints(subprocess.run(["lsof", "-ti", f"tcp:{port}", "-sTCP:LISTEN"], timeout=5, **_RUN_TEXT).stdout.strip().splitlines()) if pids: return pids with suppress(FileNotFoundError): pids.extend(int(m.group(1)) for m in re.finditer(r"pid=(\d+)", subprocess.run(["ss", "-ltnHp", f"sport = :{port}"], timeout=5, **_RUN_TEXT).stdout)) return pids def _safe_ints(tokens) -> list: out: list = [] for tok in tokens: try: out.append(int(tok)) except ValueError: pass return out def _windows_listener_pids(port: int) -> list: """PIDs in LISTENING state on ``port`` via netstat (Windows).""" from hermes_cli._subprocess_compat import windows_hide_flags result = subprocess.run(["netstat", "-ano", "-p", "TCP"], timeout=5, creationflags=windows_hide_flags(), **_RUN_TEXT) rows = (line.split() for line in result.stdout.splitlines()) return _safe_ints(p[4] for p in rows if len(p) >= 5 and p[3] == "LISTENING" and p[1].endswith(f":{port}")) def _pid_looks_like_node_bridge(pid: int) -> bool: """Fail-closed: the live process must be a ``node`` executable (a scan-time PID can be a stranger by kill time). ``_kill_port_process`` discovers PIDs from a netstat/lsof scan of a TCP port — a bare number naming a *stranger* process (#89614 class: an unverified scan-time PID force-killed later can be anything, including a critical system process). Before any kill, require the live process to actually look like our Baileys bridge: a ``node`` executable. Any ambiguity (process gone, unreadable cmdline) refuses the kill. """ try: import psutil proc = psutil.Process(pid) return "node" in (proc.name() or "").lower() or "node" in " ".join(proc.cmdline() or []).lower().split(" ", 1)[0] except Exception: return False def _kill_port_process(port: int) -> None: """Kill any node bridge *listening* on the given TCP port (never a client); SIGTERM on POSIX, taskkill /F on Windows.""" with suppress(Exception): for pid in (_windows_listener_pids(port) if _IS_WINDOWS else _listener_pids_on_port(port)): # Killing a mistyped or recycled PID is unrecoverable — verify first. if pid <= 0 or not _pid_looks_like_node_bridge(pid): logger.warning("[whatsapp] Not killing PID %s on port %d: process is not a node bridge (or identity unverifiable)", pid, port) continue if _IS_WINDOWS: from hermes_cli._subprocess_compat import windows_hide_flags # Only SubprocessError is swallowed per-PID; an OSError (e.g. taskkill missing) aborts the scan. with suppress(subprocess.SubprocessError): subprocess.run(["taskkill", "/PID", str(pid), "/F"], capture_output=True, stdin=subprocess.DEVNULL, timeout=5, creationflags=windows_hide_flags()) else: with suppress(OSError): # ProcessLookupError/PermissionError are OSError subclasses os.kill(pid, signal.SIGTERM) def _bridge_pid_is_ours(pid: int, expected_start) -> bool: """``pid`` alive AND still our bridge: kernel start time (definitive); fail closed without it. Legacy pidfiles record only the PID. The old fallback accepted a ``node`` + session-path cmdline substring as kill evidence — but a log tail, editor, or grep that merely *mentions* the session path matches that same substring (#116883), so a legacy pidfile could signal a stranger. Without a start-time fingerprint the caller must reap via the bridge-port scan instead. """ from gateway import status if not status._pid_exists(pid): return False if expected_start is None: return False return status.get_process_start_time(pid) == expected_start def _unlink_quietly(path: Path) -> None: with suppress(OSError): path.unlink() def _kill_stale_bridge_by_pidfile(session_path: Path) -> None: """Kill an orphaned bridge recorded in ``bridge.pid``, after :func:`_bridge_pid_is_ours`.""" from gateway.status import _pid_exists pid_file = session_path / "bridge.pid" if not pid_file.exists(): return pid = None recorded_start = None try: # Format: line 1 = pid, optional line 2 = kernel start time. Legacy # files written before the guard existed have only the pid. lines = pid_file.read_text(encoding="utf-8-sig").split("\n") pid = int(lines[0].strip()) if len(lines) > 1 and lines[1].strip(): recorded_start = int(lines[1].strip()) except (ValueError, OSError, TypeError, IndexError): _unlink_quietly(pid_file) return if _bridge_pid_is_ours(pid, recorded_start): with suppress(OSError): # ProcessLookupError / PermissionError included os.kill(pid, signal.SIGTERM) logger.info("[whatsapp] Killed stale bridge PID %d from pidfile", pid) elif _pid_exists(pid): reason = ("legacy pidfile lacks a start-time fingerprint and cmdline substring evidence can name a stranger" if recorded_start is None else "it is no longer the bridge (recycled onto an unrelated process)") logger.warning("[whatsapp] Not killing pidfile PID %d: %s; " "skipping to avoid killing a stranger.", pid, reason) _unlink_quietly(pid_file) def _write_bridge_pidfile(session_path: Path, pid: int) -> None: """Write the bridge PID plus its kernel start time (line 2) for identity-checked cleanup.""" with suppress(OSError): from gateway.status import get_process_start_time start = get_process_start_time(pid) (session_path / "bridge.pid").write_text(str(pid) if start is None else f"{pid}\n{start}", encoding="utf-8") def _terminate_bridge_process(proc, *, force: bool = False) -> None: """Terminate the bridge process using process-tree semantics where possible.""" action = "kill" if force else "terminate" if _IS_WINDOWS: try: result = subprocess.run( ["taskkill", "/PID", str(proc.pid), "/T"] + (["/F"] if force else []), capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=10, ) except FileNotFoundError: return getattr(proc, action)() if result.returncode != 0: raise OSError((result.stderr or result.stdout or "").strip() or f"taskkill failed for PID {proc.pid}") return import psutil with suppress(psutil.NoSuchProcess): parent = psutil.Process(proc.pid) for child in parent.children(recursive=True): with suppress(psutil.NoSuchProcess): getattr(child, action)() getattr(parent, action)() import sys sys.path.insert(0, str(Path(__file__).resolve().parents[3])) from gateway.config import Platform, PlatformConfig from gateway.platforms.whatsapp_common import WhatsAppBehaviorMixin from gateway.whatsapp_identity import normalize_whatsapp_mention_jid, to_whatsapp_jid from gateway.platforms.base import ( BasePlatformAdapter, SendResult, SUPPORTED_DOCUMENT_TYPES, cache_image_from_url, cache_audio_from_url, ) from gateway.platforms.helpers import cancel_task from gateway.platforms.event import MessageEvent, MessageType from utils import env_int def _cache_dirs() -> tuple: """``(image, audio, video, document)`` cache dirs, resolved per call so a profile override's cache matches.""" from gateway.platforms.base import get_audio_cache_dir, get_document_cache_dir, get_image_cache_dir, get_video_cache_dir return get_image_cache_dir(), get_audio_cache_dir(), get_video_cache_dir(), get_document_cache_dir() def _is_allowed_bridge_path(url: str) -> bool: """Absolute bridge path resolves (symlinks included) inside a Hermes cache dir — a rogue bridge could hand back /etc/passwd.""" try: resolved = Path(url).resolve() except (OSError, ValueError): return False for root in _cache_dirs(): with suppress(OSError, ValueError): if resolved.is_relative_to(Path(root).resolve()): return True return False def _file_content_hash(path: Path) -> str: """First 16 hex chars of SHA-256 of *path* ("" if unreadable); bridge.js reports its own as ``/health`` ``scriptHash``.""" import hashlib try: return hashlib.sha256(path.read_bytes()).hexdigest()[:16] except OSError: return "" def check_whatsapp_requirements() -> bool: """ Check if WhatsApp dependencies are available. WhatsApp requires a Node.js bridge for most implementations. """ _node = find_node_executable("node") if not _node: from pm import lazy_installs_allowed # Let connect prepare a missing runtime, but never install during discovery. return lazy_installs_allowed() try: return subprocess.run([_node, "--version"], timeout=5, env=with_hermes_node_path(), **_RUN_TEXT).returncode == 0 except Exception: return False # Env vars bridge.js consumes; injected because a multiplexed subprocess's os.environ lacks the secondary profile's .env. _BRIDGE_PASSTHROUGH_ENV = ( "WHATSAPP_ALLOWED_USERS", "WHATSAPP_ALLOW_FROM", "WHATSAPP_DM_POLICY", "WHATSAPP_GROUP_POLICY", "WHATSAPP_GROUP_ALLOWED_USERS", "WHATSAPP_GROUP_ALLOW_FROM", "WHATSAPP_REQUIRE_MENTION", "WHATSAPP_MENTION_PATTERNS", "WHATSAPP_FREE_RESPONSE_CHATS", "WHATSAPP_DEBUG", "WHATSAPP_FORWARD_OWNER_MESSAGES", "WHATSAPP_MAX_MESSAGE_LENGTH", "WHATSAPP_CHUNK_DELAY_MS", "WHATSAPP_SEND_TIMEOUT_MS", ) _TEXT_INJECT_EXTS = {".txt", ".md", ".csv", ".json", ".xml", ".yaml", ".yml", ".log", ".py", ".js", ".ts", ".html", ".css"} _MAX_TEXT_INJECT_BYTES = 100 * 1024 # matches Telegram/Discord/Slack _NATIVE_MEDIA_TYPES = {"location": MessageType.LOCATION, "live_location": MessageType.LOCATION, "sticker": MessageType.STICKER} # Inbound mediaType substring → kind; ptt = WhatsApp voice note, so "ptt" must precede "audio". _MEDIA_NEEDLES = (("image", MessageType.PHOTO), ("video", MessageType.VIDEO), ("ptt", MessageType.VOICE), ("audio", MessageType.AUDIO)) # MessageType → (bridge label, default mime); documents take their mime from SUPPORTED_DOCUMENT_TYPES instead. _MEDIA_INFO = { MessageType.PHOTO: ("image", "image/jpeg"), MessageType.VOICE: ("audio", "audio/ogg"), MessageType.AUDIO: ("audio", "audio/mpeg"), MessageType.VIDEO: ("video", "video/mp4"), MessageType.DOCUMENT: ("document", ""), } _MEDIA_TYPE_BY_BRIDGE_KIND = {"image": MessageType.PHOTO, "video": MessageType.VIDEO, "audio": MessageType.VOICE, "document": MessageType.DOCUMENT} _QUOTED_MIME_BY_BRIDGE_KIND = { "image": "image/jpeg", "video": "video/mp4", "gif": "video/mp4", "audio": "audio/ogg", "ptt": "audio/ogg", "document": "application/octet-stream", "sticker": "image/webp", } def _needs_bridge(method): """Adapter-method decorator: return ``SendResult(error=...)`` when ``_bridge_unavailable()`` says so.""" @wraps(method) async def _guarded(self, *args, **kwargs): unavailable = await self._bridge_unavailable() return SendResult(success=False, error=unavailable) if unavailable else await method(self, *args, **kwargs) return _guarded class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter): """Transport over a local Node.js (Baileys) HTTP bridge; behavior lives in ``WhatsAppBehaviorMixin``. config.extra: bridge_script / bridge_port (3000) / session_path, dm_policy / group_policy (open|allowlist|disabled|pairing), allow_from / group_allow_from, send_read_receipts.""" _DEFAULT_BRIDGE_DIR = None # resolved in __init__ splits_long_messages = True # send() chunks via truncate_message() def __init__(self, config: PlatformConfig): super().__init__(config, Platform.WHATSAPP) if WhatsAppAdapter._DEFAULT_BRIDGE_DIR is None: from gateway.platforms.whatsapp_common import resolve_whatsapp_bridge_dir WhatsAppAdapter._DEFAULT_BRIDGE_DIR = resolve_whatsapp_bridge_dir() extra = config.extra self._bridge_process: Optional[subprocess.Popen] = None self._bridge_port: int = extra.get("bridge_port", 3000) self._bridge_script: str = extra.get("bridge_script", str(self._DEFAULT_BRIDGE_DIR / "bridge.js")) self._session_path = Path(extra.get("session_path", get_hermes_dir("platforms/whatsapp/session", "whatsapp/session"))) self._reply_prefix: Optional[str] = extra.get("reply_prefix") self._dm_policy = str(_extra_or_secret(extra, "dm_policy", "WHATSAPP_DM_POLICY", "pairing")).strip().lower() self._allow_from = self._coerce_allow_list(self._select_dm_allowlist(extra, ("WHATSAPP_ALLOWED_USERS",), _wenv)) self._group_policy = str(_extra_or_secret(extra, "group_policy", "WHATSAPP_GROUP_POLICY", "pairing")).strip().lower() # Same precedence as the DM list. Until #72529 the env carrier only reached the Node bridge, so # env-only installs gated groups on an empty allowlist. _, raw_groups = self._select_allowlist( extra, ("group_allow_from", "groupAllowFrom"), ("WHATSAPP_GROUP_ALLOW_FROM", "WHATSAPP_GROUP_ALLOWED_USERS"), _wenv) self._group_allow_from = self._coerce_allow_list(raw_groups) rr = extra.get("send_read_receipts", False) self._send_read_receipts = rr if isinstance(rr, bool) else str(rr or "").strip().lower() in {"1", "true", "yes", "on"} self._mention_patterns = self._compile_mention_patterns() self._message_queue: asyncio.Queue = asyncio.Queue() self._bridge_log_fh = self._bridge_log = self._poll_task = self._http_session = None # Set by disconnect() before SIGTERMing so _check_managed_bridge_exit() can tell an intentional exit (-15/-2/0) from a crash. self._shutting_down = False # Text debounce batching: rapid bursts (forwards, paste-splits) would otherwise each trigger a separate agent turn. # Telegram cadence and ceilings (#44883); ``0`` dispatches each message immediately. self._configure_text_batch_delays() def _bridge_url(self, path: str) -> str: return f"http://127.0.0.1:{self._bridge_port}/{path}" def _bridge_req(self, method: str, path: str, timeout: float, **kwargs): """``session.`` context manager for a bridge endpoint (caller must ``async with``).""" import aiohttp return getattr(self._http_session, method)(self._bridge_url(path), **kwargs, timeout=aiohttp.ClientTimeout(total=timeout)) async def _probe_bridge_health(self) -> tuple[bool, Any]: """GET /health with a fresh session → ``(http_200, json)``; unparseable 200 body → ``(True, None)``; connection errors propagate.""" import aiohttp async with aiohttp.ClientSession() as session, session.get(self._bridge_url("health"), timeout=aiohttp.ClientTimeout(total=2)) as resp: if resp.status != 200: return False, None try: return True, await resp.json() except Exception: return True, None def _ensure_bridge_deps(self, bridge_dir: Path) -> bool: """npm install when node_modules is missing OR package.json hash != stamp file. False = fatal error set.""" _dep_stamp = bridge_dir / "node_modules" / ".hermes-pkg-hash" # holds the package.json hash of the last install _pkg_hash = _file_content_hash(bridge_dir / "package.json") try: if (bridge_dir / "node_modules").exists() and _dep_stamp.read_text(encoding="utf-8-sig").strip() == _pkg_hash and bool(_pkg_hash): return True except OSError: pass print(f"[{self.name}] Installing WhatsApp bridge dependencies...") detail = "" try: # Default 300s accommodates slow systems like an Unraid NAS. import pm _npm_bin = find_node_executable("npm") env = with_hermes_node_path() if _npm_bin is None: env = pm.ensure("npm").env installed = pm.installed_package("npm") if installed is None or installed.binary is None: raise pm.InstallError("npm", "ensured but no selected binary was recorded") _npm_bin = str(installed.binary) install_result = subprocess.run([_npm_bin, "install", "--silent"], cwd=str(bridge_dir), timeout=env_int("WHATSAPP_NPM_INSTALL_TIMEOUT", 300), env=env, **_RUN_TEXT) if install_result.returncode == 0: print(f"[{self.name}] Dependencies installed") with suppress(OSError): # Stamp is an optimization; install still succeeded if _pkg_hash: _dep_stamp.write_text(_pkg_hash, encoding="utf-8") return True print(f"[{self.name}] npm install failed: {install_result.stderr}") detail = f" ({install_result.stderr.strip()[-500:]})" if install_result.stderr else "" except Exception as e: print(f"[{self.name}] Failed to install dependencies: {e}") detail = f" ({e})" self._set_fatal_error("whatsapp_npm_install_failed", f"WhatsApp bridge npm install failed{detail}. " "Run `hermes whatsapp`, then restart `hermes gateway`.", retryable=False) return False def _attach_to_bridge(self, managed_process) -> None: import aiohttp self._bridge_process = managed_process self._http_session = aiohttp.ClientSession() self._poll_task = asyncio.create_task(self._poll_messages()) async def _reuse_running_bridge(self, bridge_path: Path) -> bool: """Adopt a connected bridge serving the on-disk bridge.js + same read-receipt config; else say why it restarts.""" try: ok, data = await self._probe_bridge_health() if not ok or data is None: return False bridge_status = data.get("status", "unknown") if bridge_status != "connected": print(f"[{self.name}] Bridge found but not connected (status: {bridge_status}), restarting") return False running_hash, disk_hash = data.get("scriptHash", ""), _file_content_hash(bridge_path) if running_hash and disk_hash and running_hash == disk_hash and bool(data.get("sendReadReceipts", False)) == self._send_read_receipts: print(f"[{self.name}] Using existing bridge (status: {bridge_status})") self._mark_connected() self._attach_to_bridge(None) # Not managed by us self._wire_plugin_handlers(None) return True stale_reason = f"running={running_hash or 'unversioned'}, disk={disk_hash}" if running_hash != disk_hash else "send_read_receipts config changed" print(f"[{self.name}] Running bridge is stale ({stale_reason}), restarting") except Exception: pass # Bridge not running, start a new one return False def _bridge_env(self) -> dict: """Subprocess env: the adapter's EFFECTIVE profile policy + profile-resolved WHATSAPP_* values + cache dirs.""" # with_hermes_node_path() copies os.environ when called with no arg: under a multiplexed secondary # that copy carries the DEFAULT profile's WHATSAPP_* values, so every bridge-consumed key is # re-resolved from this profile (dropped on a scoped miss), never inherited from the launch env. bridge_env = with_hermes_node_path() reply_prefix = _wenv("WHATSAPP_REPLY_PREFIX") if reply_prefix: bridge_env["WHATSAPP_REPLY_PREFIX"] = reply_prefix elif self._reply_prefix is not None: bridge_env["WHATSAPP_REPLY_PREFIX"] = self._reply_prefix else: bridge_env.pop("WHATSAPP_REPLY_PREFIX", None) bridge_env["WHATSAPP_SEND_READ_RECEIPTS"] = "true" if self._send_read_receipts else "false" for _key, _v in [("WHATSAPP_MODE", _wenv("WHATSAPP_MODE", "self-chat"))] + [(k, _wenv(k)) for k in _BRIDGE_PASSTHROUGH_ENV]: if _v: bridge_env[_key] = _v else: bridge_env.pop(_key, None) # bridge.js gates DMs BEFORE Python sees them: it must run the same dm_policy / allow_from the # adapter resolved (scoped env → this profile's YAML → default), or a secondary's YAML # ``dm_policy: pairing`` runs under the default profile's allowlist and drops valid pairing DMs. bridge_env["WHATSAPP_DM_POLICY"] = self._dm_policy bridge_env["WHATSAPP_GROUP_POLICY"] = self._group_policy for env_key, ids in (("WHATSAPP_ALLOWED_USERS", self._allow_from), ("WHATSAPP_GROUP_ALLOWED_USERS", self._group_allow_from)): if ids: bridge_env[env_key] = ",".join(sorted(ids)) else: bridge_env.pop(env_key, None) # Without these the bridge hardcodes ~/.hermes/{image,audio,document}_cache (wrong under HERMES_HOME/profiles/cache layout). img_dir, audio_dir, _video_dir, doc_dir = _cache_dirs() bridge_env.update(HERMES_IMAGE_CACHE_DIR=str(img_dir), HERMES_AUDIO_CACHE_DIR=str(audio_dir), HERMES_DOCUMENT_CACHE_DIR=str(doc_dir)) return bridge_env def _bridge_died(self, detail: str) -> bool: print(f"[{self.name}] {detail}") print(f"[{self.name}] Check log: {self._bridge_log}") self._close_bridge_log() return False async def _poll_bridge_health(self, died_msg: str) -> tuple[Optional[bool], bool, dict]: """Poll /health up to 15×1s → ``(connected, http_ready, data)``; connected False = process died (reported), None = timeout.""" http_ready = False data: dict = {} for attempt in range(15): await asyncio.sleep(1) if self._bridge_process.poll() is not None: return self._bridge_died(died_msg.format(code=self._bridge_process.returncode)), http_ready, data try: ok, d = await self._probe_bridge_health() if ok: http_ready = True if d is not None: data = d if data.get("status") == "connected": print(f"[{self.name}] Bridge ready (status: connected)") return True, http_ready, data except Exception: continue return None, http_ready, data async def _wait_for_bridge(self) -> bool: """Phase 1: HTTP up (≤15s). Phase 2: ``status: connected`` (≤15s more; warns but proceeds if still connecting).""" connected, http_ready, data = await self._poll_bridge_health("Bridge process died (exit code {code})") if connected is False: return False if not http_ready: return self._bridge_died("Bridge HTTP server did not start in 15s") if data.get("status") != "connected": print(f"[{self.name}] Bridge HTTP ready, waiting for WhatsApp connection...") connected, _, _ = await self._poll_bridge_health("Bridge process died during connection") if connected is False: return False if connected is None: print(f"[{self.name}] ⚠ WhatsApp not connected after 30s") print(f"[{self.name}] Bridge log: {self._bridge_log}") print(f"[{self.name}] If session expired, re-pair: hermes whatsapp") return True def _preflight(self) -> bool: """Node + bridge script + creds.json present, else a non-retryable fatal error (an unpaired bridge only prints QR codes; retries would pay 30s each).""" bridge_path = Path(self._bridge_script) creds_path = self._session_path / "creds.json" checks = ( (check_whatsapp_requirements, ("[%s] Node.js not found. WhatsApp requires Node.js.", self.name), "whatsapp_node_missing", "Node.js is not installed — install Node.js and re-run `hermes gateway`."), (bridge_path.exists, ("[%s] Bridge script not found: %s", self.name, bridge_path), "whatsapp_bridge_missing", f"WhatsApp bridge script missing at {bridge_path}."), (creds_path.exists, ("[%s] WhatsApp is enabled but not paired (no creds.json at %s). Pair from the dashboard or run " "`hermes whatsapp`; remove WHATSAPP_ENABLED from your .env to disable.", self.name, creds_path), "whatsapp_not_paired", "WhatsApp enabled but not paired — pair from the dashboard or run `hermes whatsapp`."), ) for ok, warn_args, code, message in checks: if not ok(): logger.warning(*warn_args) self._set_fatal_error(code, message, retryable=False) return False logger.info("[%s] Bridge found at %s", self.name, bridge_path) return True async def connect(self, *, is_reconnect: bool = False) -> bool: """Start (or adopt) the Node.js bridge and wait for it to be ready.""" if find_node_executable("node") is None: import pm try: await asyncio.to_thread(pm.ensure, "node") except pm.InstallError as exc: self._set_fatal_error("whatsapp_node_missing", str(exc), retryable=False) return False if not self._preflight(): return False bridge_path = Path(self._bridge_script) lock_acquired = False try: if not self._acquire_platform_lock('whatsapp-session', str(self._session_path), 'WhatsApp session'): return False lock_acquired = True except Exception as e: logger.warning("[%s] Could not acquire session lock (non-fatal): %s", self.name, e) try: if not self._ensure_bridge_deps(bridge_path.parent): return False self._session_path.mkdir(parents=True, exist_ok=True) if await self._reuse_running_bridge(bridge_path): return True _kill_stale_bridge_by_pidfile(self._session_path) _kill_port_process(self._bridge_port) await asyncio.sleep(1) # Bridge output goes to a log file so QR codes, errors, and reconnection messages survive for troubleshooting. self._bridge_log = self._session_path.parent / "bridge.log" self._bridge_log_fh = bridge_log_fh = open(self._bridge_log, "a", encoding="utf-8") node = find_node_executable("node") if node is None: raise RuntimeError("Node.js is no longer available; run `hermes pm install`") self._bridge_process = subprocess.Popen( [node, str(bridge_path), "--port", str(self._bridge_port), "--session", str(self._session_path), "--mode", _wenv("WHATSAPP_MODE", "self-chat")], stdout=bridge_log_fh, stderr=bridge_log_fh, env=self._bridge_env(), **windows_detach_popen_kwargs()) _write_bridge_pidfile(self._session_path, self._bridge_process.pid) if not await self._wait_for_bridge(): return False self._attach_to_bridge(self._bridge_process) self._mark_connected() print(f"[{self.name}] Bridge started on port {self._bridge_port}") self._wire_plugin_handlers(None) return True except Exception as e: logger.error("[%s] Failed to start bridge: %s", self.name, e, exc_info=True) return False finally: if not self._running: if lock_acquired: self._release_platform_lock() self._close_bridge_log() def _close_bridge_log(self) -> None: if self._bridge_log_fh: with suppress(Exception): self._bridge_log_fh.close() self._bridge_log_fh = None async def _check_managed_bridge_exit(self) -> Optional[str]: returncode = self._bridge_process.poll() if self._bridge_process is not None else None if returncode is None: return None # getattr-with-default: tests build the adapter via ``__new__`` without __init__. # A container stop (e.g. s6-overlay's stage-3 broadcast) reaches the bridge child (its own session) before the # gateway's stop flow reaches this adapter's disconnect(), so ``_shutting_down`` alone # misses that window and a normal -15/-2 exit reads as a fatal crash (#127047). The # runner flips ``_stop_requested_by_signal`` in its signal handler — the first thing # that runs on any signal-driven stop — so consult it too. Un-guarded it stays fatal: # a bridge killed while the gateway keeps running still needs the reconnect watcher. runner_stop = getattr(getattr(self, "gateway_runner", None), "_stop_requested_by_signal", False) if (getattr(self, "_shutting_down", False) or runner_stop) and returncode in {0, -2, -15}: logger.info("[%s] Bridge exited during shutdown (code %d).", self.name, returncode) return None message = f"WhatsApp bridge process exited unexpectedly (code {returncode})." if not self.has_fatal_error: logger.error("[%s] %s", self.name, message) self._set_fatal_error("whatsapp_bridge_exited", message, retryable=True) self._close_bridge_log() await self._notify_fatal_error() return self.fatal_error_message or message def _terminate_bridge(self, *, force: bool) -> None: try: _terminate_bridge_process(self._bridge_process, force=force) except (ProcessLookupError, PermissionError): getattr(self._bridge_process, "kill" if force else "terminate")() async def disconnect(self) -> None: """Stop the WhatsApp bridge and clean up any orphaned processes.""" self._shutting_down = True # flip BEFORE signalling so send()/poll loop don't report the intentional exit as fatal if not self._bridge_process: print(f"[{self.name}] Disconnecting (external bridge left running)") else: try: self._terminate_bridge(force=False) await asyncio.sleep(1) if self._bridge_process.poll() is None: self._terminate_bridge(force=True) except Exception as e: print(f"[{self.name}] Error stopping bridge: {e}") _unlink_quietly(self._session_path / "bridge.pid") await cancel_task(self._poll_task) if self._http_session and not self._http_session.closed: await self._http_session.close() self._poll_task = self._http_session = self._bridge_process = None self._release_platform_lock() self._mark_disconnected() self._close_bridge_log() print(f"[{self.name}] Disconnected") async def _bridge_unavailable(self) -> Optional[str]: return "Not connected" if not self._running or not self._http_session else (await self._check_managed_bridge_exit() or None) async def _post_bridge_message(self, path: str, payload: Dict[str, Any], *, timeout: float) -> SendResult: """POST to the bridge; 200 → SendResult(messageId, raw_response), else the error text.""" try: async with self._bridge_req("post", path, timeout, json=payload) as resp: if resp.status == 200: data = await resp.json() return SendResult(success=True, message_id=data.get("messageId"), raw_response=data) return SendResult(success=False, error=await resp.text()) except Exception as e: return SendResult(success=False, error=str(e)) @_needs_bridge async def send(self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: """Format markdown for WhatsApp, chunk preserving code blocks, send sequentially.""" if not content or not content.strip(): return SendResult(success=True, message_id=None) chat_id = to_whatsapp_jid(chat_id) try: chunks = self.truncate_message(self.format_message(content), self._outgoing_chunk_limit()) sent_message_ids: list[str] = [] last_message_id = None for idx, chunk in enumerate(chunks): payload: Dict[str, Any] = {"chatId": chat_id, "message": chunk} if reply_to and idx == 0: payload["replyTo"] = reply_to # Reply-to on the first chunk only. result = await self._post_bridge_message("send", payload, timeout=30) if not result.success: return SendResult(success=False, error=result.error) last_message_id = result.message_id if last_message_id: sent_message_ids.append(str(last_message_id)) if len(chunks) > 1: await asyncio.sleep(0.3) # avoid rate limiting between chunks return SendResult(success=True, message_id=last_message_id, continuation_message_ids=tuple(sent_message_ids[:-1]), raw_response={"message_ids": sent_message_ids}) except Exception as e: return SendResult(success=False, error=str(e)) @_needs_bridge async def edit_message(self, chat_id: str, message_id: str, content: str, *, finalize: bool = False) -> SendResult: try: async with self._bridge_req("post", "edit", 15, json={"chatId": to_whatsapp_jid(chat_id), "messageId": message_id, "message": content}) as resp: return SendResult(success=True, message_id=message_id) if resp.status == 200 else SendResult(success=False, error=await resp.text()) except Exception as e: return SendResult(success=False, error=str(e)) @_needs_bridge async def _send_media_to_bridge(self, chat_id: str, file_path: str, media_type: str, caption: Optional[str] = None, file_name: Optional[str] = None) -> SendResult: if not os.path.exists(file_path): return SendResult(success=False, error=f"File not found: {file_path}") jid = to_whatsapp_jid(chat_id) payload: Dict[str, Any] = {"chatId": jid, "filePath": file_path, "mediaType": media_type} payload.update({k: v for k, v in (("caption", caption), ("fileName", file_name)) if v}) result = await self._post_bridge_message("send-media", payload, timeout=120) if result.success and result.message_id: # A later quote of this attachment carries only a thumbnail stub; the bridge's cache # knows inbound media only, so index our own sends (the cron-delivered image case). from gateway import rich_sent_store mime = mimetypes.guess_type(file_path)[0] or _MEDIA_INFO.get(_MEDIA_TYPE_BY_BRIDGE_KIND.get(media_type), ("", ""))[1] await rich_sent_store.record_media_async(jid, result.message_id, [(file_path, mime or "application/octet-stream")]) return result @_needs_bridge async def send_poll(self, chat_id: str, question: str, options: list[str], *, selectable_count: int = 1) -> SendResult: """Native WhatsApp poll (low-level transport primitive; approval UX stays gateway-owned).""" payload: Dict[str, Any] = {"chatId": to_whatsapp_jid(chat_id), "question": question, "options": list(options or []), "selectableCount": selectable_count} return await self._post_bridge_message("send-poll", payload, timeout=30) async def send_clarify(self, chat_id: str, question: str, choices: Optional[list], clarify_id: str, session_key: str, metadata: Optional[Dict[str, Any]] = None) -> SendResult: """Multiple-choice clarify as a native poll (the pick arrives as message text for the normal intercept); else text prompt.""" clean_choices = [str(choice).strip() for choice in (choices or []) if str(choice).strip()] if 2 <= len(clean_choices) <= 12: result = await self.send_poll(chat_id, str(question or "").strip(), clean_choices, selectable_count=1) if result.success: return result logger.warning("[%s] Native WhatsApp clarify poll failed; falling back to text: %s", self.name, result.error) return await super().send_clarify(chat_id=chat_id, question=question, choices=choices, clarify_id=clarify_id, session_key=session_key, metadata=metadata) @_needs_bridge async def send_location(self, chat_id: str, latitude: float, longitude: float, *, name: Optional[str] = None, address: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: try: payload: Dict[str, Any] = {"chatId": to_whatsapp_jid(chat_id), "latitude": float(latitude), "longitude": float(longitude)} except Exception as e: return SendResult(success=False, error=str(e)) payload.update({k: v for k, v in (("name", name), ("address", address)) if v}) return await self._post_bridge_message("send-location", payload, timeout=30) async def send_image(self, chat_id: str, image_url: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> SendResult: """Download image URL to cache, send natively via bridge (``metadata`` honors the base contract).""" try: local_path = await cache_image_from_url(image_url) return await self._send_media_to_bridge(chat_id, local_path, "image", caption) except Exception: return await super().send_image(chat_id, image_url, caption, reply_to, metadata) async def send_image_file(self, chat_id: str, image_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs) -> SendResult: return await self._send_media_to_bridge(chat_id, image_path, "image", caption) async def send_video(self, chat_id: str, video_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs) -> SendResult: return await self._send_media_to_bridge(chat_id, video_path, "video", caption) async def send_voice(self, chat_id: str, audio_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, **kwargs) -> SendResult: return await self._send_media_to_bridge(chat_id, audio_path, "audio", caption) async def send_document(self, chat_id: str, file_path: str, caption: Optional[str] = None, file_name: Optional[str] = None, reply_to: Optional[str] = None, **kwargs) -> SendResult: return await self._send_media_to_bridge(chat_id, file_path, "document", caption, file_name or os.path.basename(file_path)) async def send_typing(self, chat_id: str, metadata=None) -> None: if await self._bridge_unavailable(): return with suppress(Exception): import aiohttp # ``async with`` — a bare ``await session.post(...)`` leaves the response (and its CLOSE_WAIT socket) alive until GC. async with self._http_session.post(self._bridge_url("typing"), json={"chatId": to_whatsapp_jid(chat_id)}, timeout=aiohttp.ClientTimeout(total=5)): pass async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: if not self._running or not self._http_session: return {"name": "Unknown", "type": "dm"} if not await self._check_managed_bridge_exit(): try: async with self._bridge_req("get", f"chat/{to_whatsapp_jid(chat_id)}", 10) as resp: if resp.status == 200: data = await resp.json() return {"name": data.get("name", chat_id), "type": "group" if data.get("isGroup") else "dm", "participants": data.get("participants", [])} except Exception as e: logger.debug("Could not get WhatsApp chat info for %s: %s", chat_id, e) return {"name": chat_id, "type": "dm"} async def _report_bridge_exit(self) -> bool: bridge_exit = await self._check_managed_bridge_exit() if bridge_exit: print(f"[{self.name}] {bridge_exit}") return bool(bridge_exit) async def _poll_messages(self) -> None: while self._running: if not self._http_session or await self._report_bridge_exit(): break try: async with self._bridge_req("get", "messages", 30) as resp: if resp.status == 200: for msg_data in await resp.json(): event = await self._build_message_event(msg_data) if event: # Fire-and-forget: a slow bridge /read must not delay dispatch. asyncio.create_task(self._send_read_receipt(msg_data)) if event.message_type == MessageType.TEXT: self._enqueue_text_event(event) else: await self.handle_message(event) except asyncio.CancelledError: break except Exception as e: if await self._report_bridge_exit(): break print(f"[{self.name}] Poll error: {e}") await asyncio.sleep(5) await asyncio.sleep(1) # Poll interval async def _send_read_receipt(self, data: Dict[str, Any]) -> None: key = data.get("readReceiptKey") if not self._send_read_receipts or not self._http_session or not isinstance(key, dict): return try: async with self._bridge_req("post", "read", 5, json={"key": key}) as resp: if resp.status != 200: logger.warning("[%s] WhatsApp read receipt failed with HTTP %s", self.name, resp.status) except Exception as exc: logger.warning("[%s] WhatsApp read receipt failed: %s", self.name, exc) _SPLIT_THRESHOLD = 6000 # WhatsApp supports ~65K chars; generous threshold @staticmethod def _classify_bridge_message(data: Dict[str, Any]) -> MessageType: media_type = str(data.get("mediaType", "") or "") if media_type in _NATIVE_MEDIA_TYPES: return _NATIVE_MEDIA_TYPES[media_type] if not data.get("hasMedia"): return MessageType.TEXT return next((kind for needle, kind in _MEDIA_NEEDLES if needle in media_type), MessageType.DOCUMENT) async def _collect_bridge_media(self, data: Dict[str, Any], msg_type: MessageType) -> tuple[list, list]: """``mediaUrls`` → ``(cached_urls, media_types)``: remote image/audio cached locally; absolute paths only inside a cache dir.""" accepted: list[tuple] = [] # (url_or_path, mime) label, default_mime = _MEDIA_INFO.get(msg_type, (None, "")) bridge_mime = str(data.get("mime") or "").strip() for url in data.get("mediaUrls", []): mime = bridge_mime or (SUPPORTED_DOCUMENT_TYPES.get(Path(url).suffix.lower(), "application/octet-stream") if msg_type == MessageType.DOCUMENT else default_mime) if url.startswith(("http://", "https://")) and msg_type in {MessageType.PHOTO, MessageType.VOICE, MessageType.AUDIO}: cacher, ext = (cache_image_from_url, ".jpg") if msg_type == MessageType.PHOTO else (cache_audio_from_url, ".ogg") try: url = await cacher(url, ext=ext) print(f"[{self.name}] Cached user {label}: {url}", flush=True) except Exception as e: print(f"[{self.name}] Failed to cache {label}: {e}", flush=True) accepted.append((url, mime)) elif label is not None and os.path.isabs(url): if _is_allowed_bridge_path(url): accepted.append((url, mime)) print(f"[{self.name}] Using bridge-cached {label}: {url}", flush=True) else: print(f"[{self.name}] Rejected bridge {label} path outside cache dir: {url}", flush=True) else: accepted.append((url, "unknown")) return [u for u, _ in accepted], [m for _, m in accepted] def _inject_document_text(self, cached_urls: list, body: str) -> tuple[str, list[bool]]: """Prepend text-readable document contents (≤100KB) so the agent reads them inline; returns ``(body, media_text_inlined)`` with one flag per ``cached_urls`` entry (True = injected).""" inlined = [False] * len(cached_urls) for i, doc_path in enumerate(cached_urls): p = Path(doc_path) if p.suffix.lower() not in _TEXT_INJECT_EXTS: continue try: file_size = p.stat().st_size if file_size > _MAX_TEXT_INJECT_BYTES: print(f"[{self.name}] Skipping text injection for {doc_path} ({file_size} bytes > {_MAX_TEXT_INJECT_BYTES})", flush=True) continue content = p.read_text(encoding="utf-8-sig", errors="replace") parts = p.name.split("_", 2) # strip the doc__ prefix for display injection = f"[Content of {parts[2] if len(parts) >= 3 else p.name}]:\n{content}" body = f"{injection}\n\n{body}" if body else injection inlined[i] = True print(f"[{self.name}] Injected text content from: {doc_path}", flush=True) except Exception as e: print(f"[{self.name}] Failed to read document text: {e}", flush=True) return body, inlined def _quoted_media(self, data: Dict[str, Any], raw_reply_id: Any) -> list[tuple[str, str]]: """``(path, mime)`` for the quoted message's attachment, folded into this event's own media so the vision/audio pipeline sees it like a direct send. ``contextInfo.quotedMessage`` carries only a thumbnail stub, so the bridge resolves INBOUND quotes from its download cache (``quotedMediaUrls``, guarded like direct media: a rogue bridge could hand back /etc/passwd); quotes of OUR media (cron chart, generated image — any path we chose) resolve from the outbound index written by ``_send_media_to_bridge``.""" quoted_type = str(data.get("quotedMediaType") or "").strip() accepted: list[tuple[str, str]] = [] for path in data.get("quotedMediaUrls") or []: if not (isinstance(path, str) and os.path.isabs(path) and _is_allowed_bridge_path(path)): print(f"[{self.name}] Rejected quoted-media path outside cache dir: {path}", flush=True) continue accepted.append((path, _QUOTED_MIME_BY_BRIDGE_KIND.get(quoted_type, "application/octet-stream"))) if not accepted and raw_reply_id is not None: from gateway import rich_sent_store accepted = rich_sent_store.lookup_media(data.get("chatId", ""), str(raw_reply_id)) for path, _ in accepted: print(f"[{self.name}] Attached quoted-reply media: {path}", flush=True) return accepted async def _build_message_event(self, data: Dict[str, Any]) -> Optional[MessageEvent]: """Build a MessageEvent from bridge message data, downloading images to cache.""" try: if not self._should_process_message(data): return None msg_type = self._classify_bridge_message(data) source = self.build_source(chat_id=data.get("chatId", ""), chat_name=data.get("chatName"), chat_type="group" if data.get("isGroup", False) else "dm", user_id=data.get("senderId"), user_name=data.get("senderName"), message_id=data.get("messageId")) cached_urls, media_types = await self._collect_bridge_media(data, msg_type) body = data.get("body", "") if data.get("isGroup"): body = self._clean_bot_mention_text(body, data) if msg_type == MessageType.VOICE and cached_urls and str(body).strip().lower() == "[ptt received]": body = "" # Bridge placeholder for captionless voice notes; the audio is the payload. # Quoted message stays in structured fields only — GatewayRunner renders the "[Replying to: ...]" pointer. quoted = bool(data.get("hasQuotedMessage")) raw_reply_id = data.get("quotedMessageId") if quoted else None if quoted: for path, mime in self._quoted_media(data, raw_reply_id): cached_urls.append(path) media_types.append(mime) media_text_inlined: list[bool] = [] if msg_type == MessageType.DOCUMENT and cached_urls: body, media_text_inlined = self._inject_document_text(cached_urls, body) native_metadata = data.get("nativeMetadata") metadata: Dict[str, Any] = {k: v for k, v in ( ("whatsapp_native_type", str(data.get("nativeType") or "").strip()), ("whatsapp_native", native_metadata if isinstance(native_metadata, dict) else None), ) if v} # ``fromOwner`` = owner-typed inbound fromMe (gated by WHATSAPP_FORWARD_OWNER_MESSAGES at the bridge); surfaced as # metadata AND a text prefix so the marker survives downstream failures before silent_ingest. if data.get("fromOwner"): metadata["whatsapp_from_owner"] = True if not body.startswith(_OWNER_REPLY_PREFIX): body = f"{_OWNER_REPLY_PREFIX}{body}" return MessageEvent( text=body, message_type=msg_type, source=source, raw_message=data, message_id=data.get("messageId"), media_urls=cached_urls, media_types=media_types, media_text_inlined=media_text_inlined, metadata=metadata, reply_to_message_id=str(raw_reply_id) if raw_reply_id is not None else None, reply_to_text=str(data.get("quotedText") or "").strip() or None, reply_to_author_id=(self._normalize_whatsapp_id(data.get("quotedParticipant")) or None) if quoted else None, reply_to_is_own_message=self._message_is_reply_to_bot(data) if quoted else False, ) except Exception as e: print(f"[{self.name}] Error building event: {e}") return None # ── Plugin glue: register(ctx) plus the hooks for gateway/run.py, gateway/config.py, hermes_cli/gateway.py, send_message_tool.py. _WA_EXT_MEDIA_TYPE = { **dict.fromkeys((".jpg", ".jpeg", ".png", ".webp", ".gif"), "image"), **dict.fromkeys((".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"), "video"), **dict.fromkeys((".ogg", ".opus", ".mp3", ".wav", ".m4a", ".flac"), "audio"), } def _bridge_media_type(file_path: str, is_voice: bool, force_document: bool) -> str: """Local file → bridge ``mediaType`` (image|video|audio|document); ``force_document`` = the [[as_document]] directive.""" return "document" if force_document else "audio" if is_voice else _WA_EXT_MEDIA_TYPE.get(os.path.splitext(file_path)[1].lower(), "document") def _normalize_outbound_mentions(mentions: list[str] | None) -> list[str]: """Valid participant JIDs, deduplicated, order preserved.""" return list(dict.fromkeys(jid for m in mentions or () if (jid := normalize_whatsapp_mention_jid(m)))) async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False, caption=None, mentions=None): """Out-of-process delivery via the bridge HTTP API (standalone_sender_fn: cron apart from the gateway); ``caption`` rides on the media bubble.""" try: import aiohttp except ImportError: return send_error("aiohttp not installed. Run: pip install aiohttp") try: bridge_port = (getattr(pconfig, "extra", {}) or {}).get("bridge_port", 3000) normalized_chat_id = to_whatsapp_jid(chat_id) media = media_files or [] # A caption only applies to a single media file — never repeat it across a multi-file send. media_caption = caption if (caption and len(media) == 1) else None pending_mentions = _normalize_outbound_mentions(mentions) def _mention_first_payload(payload): nonlocal pending_mentions if pending_mentions: payload["mentions"] = pending_mentions pending_mentions = [] return payload last_message_id = None async with aiohttp.ClientSession() as session: if pending_mentions: async with session.get( f"http://localhost:{bridge_port}/health", timeout=aiohttp.ClientTimeout(total=5), ) as resp: health = await resp.json() if resp.status == 200 else {} if not (health.get("capabilities") or {}).get("outboundMentions"): return {"error": ( "WhatsApp bridge does not support native mentions; " "restart it from the same Hermes version.")} async def _post(path, payload, total, error_label=None): """``(messageId, None)`` on 200, else ``(None, error_dict)`` (body read only when labelled).""" url = f"http://localhost:{bridge_port}/{path}" async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=total)) as resp: if resp.status == 200: return (await resp.json()).get("messageId"), None return None, {} if error_label is None else send_error(f"WhatsApp {error_label} error ({resp.status}): {await resp.text()}") # 1) Text first (skipped when media-only or when the text rides as the caption). if (message or "").strip() and not media_caption: payload = _mention_first_payload({"chatId": normalized_chat_id, "message": message}) last_message_id, err = await _post("send", payload, 30, "bridge") if err: return err # 2) Each media file as a native attachment (mediaType picks the WhatsApp kind). for media_path, is_voice in media: if not os.path.exists(media_path): # In caption mode the words would vanish with the missing file — deliver the caption as a plain message. if media_caption: try: payload = _mention_first_payload({"chatId": normalized_chat_id, "message": media_caption}) await _post("send", payload, 30) except Exception: logger.warning("WhatsApp caption-fallback send failed for missing media") return send_error(f"WhatsApp media file not found: {media_path}") media_type = _bridge_media_type(media_path, is_voice, force_document) payload: Dict[str, Any] = {"chatId": normalized_chat_id, "filePath": media_path, "mediaType": media_type} payload.update({k: v for k, v in (("fileName", os.path.basename(media_path) if media_type == "document" else None), ("caption", media_caption)) if v}) payload = _mention_first_payload(payload) mid, err = await _post("send-media", payload, 120, "media") if err: return err last_message_id = mid or last_message_id return {"success": True, "platform": "whatsapp", "chat_id": normalized_chat_id, "message_id": last_message_id} except Exception as e: return send_error(f"WhatsApp send failed: {e}") def interactive_setup() -> None: """Guide the user through WhatsApp setup (CLI helpers lazy-imported).""" from hermes_cli.config import get_env_value, remove_env_value, save_env_value from hermes_cli.cli_output import prompt, prompt_yes_no, print_header, print_info, print_success print_header("WhatsApp") print_info("WhatsApp uses a local Node.js bridge (WhatsApp Web client).") print_info("Start the bridge separately; the gateway connects to it over HTTP.") if (get_env_value("WHATSAPP_ENABLED") or "").lower() in {"true", "1", "yes"}: print_info("WhatsApp: already enabled") if not prompt_yes_no("Reconfigure WhatsApp?", False): return if not prompt_yes_no("Enable WhatsApp?", True): save_env_value("WHATSAPP_ENABLED", "false") print_info("WhatsApp left disabled") return save_env_value("WHATSAPP_ENABLED", "true") print_success("WhatsApp enabled") allowed_users = prompt("Allowed user IDs (comma-separated, leave empty for no allowlist)") if allowed_users: save_env_value("WHATSAPP_ALLOWED_USERS", allowed_users.replace(" ", "")) print_success("WhatsApp allowlist configured") home_channel = prompt("Home chat ID for cron delivery (leave empty to skip)").strip() if home_channel: save_env_value("WHATSAPP_HOME_CHANNEL", home_channel) elif remove_env_value("WHATSAPP_HOME_CHANNEL"): print_info("Home channel cleared.") # config.yaml whatsapp: key → env var. Env vars take precedence over YAML. _YAML_BRIDGE = ( # (yaml key, env var, kind) for apply_yaml_bridge ("require_mention", "WHATSAPP_REQUIRE_MENTION", "lower"), ("dm_policy", "WHATSAPP_DM_POLICY", "lower"), ("group_policy", "WHATSAPP_GROUP_POLICY", "lower"), ("mention_patterns", "WHATSAPP_MENTION_PATTERNS", "json"), ("free_response_chats", "WHATSAPP_FREE_RESPONSE_CHATS", "csv"), ("allow_from", "WHATSAPP_ALLOWED_USERS", "csv"), ("group_allow_from", "WHATSAPP_GROUP_ALLOWED_USERS", "csv"), ) def _apply_yaml_config(yaml_cfg: dict, whatsapp_cfg: dict) -> dict | None: """``apply_yaml_config_fn`` (#24849): config.yaml whatsapp: keys → WHATSAPP_* env (env wins; skipped under a multiplexed secondary profile's scope, #80099) + ``PlatformConfig.extra`` (extra-first readers).""" return _apply_yaml_bridge(whatsapp_cfg, _YAML_BRIDGE) def _is_connected(config) -> bool: """Connected == WHATSAPP_ENABLED opt-in (or an enabled PlatformConfig with extras); auth lives in the bridge.""" if config is not None and getattr(config, "enabled", False) and (getattr(config, "extra", {}) or {}): return True # Via hermes_cli.gateway.get_env_value (not os.getenv) so setup-status callers that patch it observe the same value. import hermes_cli.gateway as gateway_mod return (gateway_mod.get_env_value("WHATSAPP_ENABLED") or "").strip().lower() in {"true", "1", "yes"} def register(ctx) -> None: ctx.register_platform( name="whatsapp", label="WhatsApp", adapter_factory=WhatsAppAdapter, check_fn=check_whatsapp_requirements, is_connected=_is_connected, required_env=["WHATSAPP_ENABLED"], install_hint="WhatsApp requires a Node.js bridge — see the WhatsApp messaging docs", setup_fn=interactive_setup, apply_yaml_config_fn=_apply_yaml_config, allowed_users_env="WHATSAPP_ALLOWED_USERS", allow_all_env="WHATSAPP_ALLOW_ALL_USERS", cron_deliver_env_var="WHATSAPP_HOME_CHANNEL", standalone_sender_fn=_standalone_send, max_message_length=4096, emoji="💬", allow_update_command=True, )