| 1 | """ |
| 2 | WhatsApp bridge subprocess manager. |
| 3 | |
| 4 | No agent/tool dependencies. |
| 5 | """ |
| 6 | |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import asyncio |
| 10 | import hashlib |
| 11 | import json |
| 12 | import os |
| 13 | import platform |
| 14 | import shutil |
| 15 | import subprocess |
| 16 | import threading |
| 17 | from collections import deque |
| 18 | from pathlib import Path |
| 19 | from typing import Any, Sequence |
| 20 | |
| 21 | from helpers.print_style import PrintStyle |
| 22 | |
| 23 | |
| 24 | _bridge_lock: asyncio.Lock | None = None |
| 25 | _bridge_lock_loop: asyncio.AbstractEventLoop | None = None |
| 26 | _bridge_config: dict = {} # config the running bridge was started with |
| 27 | |
| 28 | MAX_STARTUP_LOG_LINES = 80 |
| 29 | STARTUP_WAIT_ATTEMPTS = 20 |
| 30 | STARTUP_WAIT_SECONDS = 0.5 |
| 31 | DEPENDENCY_FAILURE_MARKERS = ( |
| 32 | "ERR_MODULE_NOT_FOUND", |
| 33 | "MODULE_NOT_FOUND", |
| 34 | "Cannot find module", |
| 35 | "Cannot find package", |
| 36 | ) |
| 37 | |
| 38 | |
| 39 | # ------------------------------------------------------------------ |
| 40 | # Process wrapper with destructor |
| 41 | # ------------------------------------------------------------------ |
| 42 | |
| 43 | class _BridgeProcess: |
| 44 | """Thin wrapper around Popen — kills the process on garbage collection.""" |
| 45 | |
| 46 | def __init__(self, process: subprocess.Popen, port: int): |
| 47 | self._process = process |
| 48 | self._port = port |
| 49 | self._recent_output: deque[str] = deque(maxlen=MAX_STARTUP_LOG_LINES) |
| 50 | |
| 51 | def poll(self) -> int | None: |
| 52 | return self._process.poll() |
| 53 | |
| 54 | def terminate(self) -> None: |
| 55 | self._process.terminate() |
| 56 | |
| 57 | def wait(self, timeout: float | None = None) -> int: |
| 58 | return self._process.wait(timeout=timeout) |
| 59 | |
| 60 | def kill(self) -> None: |
| 61 | self._process.kill() |
| 62 | |
| 63 | def remember_output(self, text: str) -> None: |
| 64 | self._recent_output.append(text) |
| 65 | |
| 66 | def recent_output(self) -> str: |
| 67 | return "\n".join(self._recent_output) |
| 68 | |
| 69 | @property |
| 70 | def stdout(self): |
| 71 | return self._process.stdout |
| 72 | |
| 73 | def __del__(self) -> None: |
| 74 | try: |
| 75 | if self._process.poll() is None: |
| 76 | PrintStyle.error("WhatsApp: bridge still running on GC, killing") |
| 77 | self._process.terminate() |
| 78 | try: |
| 79 | self._process.wait(timeout=5) |
| 80 | except subprocess.TimeoutExpired: |
| 81 | self._process.kill() |
| 82 | _kill_port_process(self._port) |
| 83 | except Exception as e: |
| 84 | PrintStyle.error(f"WhatsApp: bridge destructor error: {e}") |
| 85 | |
| 86 | |
| 87 | _bridge_process: _BridgeProcess | None = None |
| 88 | |
| 89 | REPO_ROOT = Path(__file__).resolve().parents[3] |
| 90 | BRIDGE_DIR = str(Path(__file__).parent.parent / "whatsapp-bridge") |
| 91 | BRIDGE_SCRIPT = os.path.join(BRIDGE_DIR, "bridge.js") |
| 92 | BRIDGE_PACKAGE_JSON = os.path.join(BRIDGE_DIR, "package.json") |
| 93 | BRIDGE_PACKAGE_LOCK = os.path.join(BRIDGE_DIR, "package-lock.json") |
| 94 | BRIDGE_RUNTIME_DIR = os.path.join(REPO_ROOT, "usr", "whatsapp", "bridge-runtime") |
| 95 | BRIDGE_INSTALL_STATE = os.path.join(BRIDGE_RUNTIME_DIR, "deps-state.json") |
| 96 | BRIDGE_NPM_CACHE = os.path.join(BRIDGE_RUNTIME_DIR, "npm-cache") |
| 97 | NODE_MODULES_DIR = os.path.join(BRIDGE_DIR, "node_modules") |
| 98 | |
| 99 | |
| 100 | # ------------------------------------------------------------------ |
| 101 | # Public API |
| 102 | # ------------------------------------------------------------------ |
| 103 | |
| 104 | async def start_bridge( |
| 105 | port: int, |
| 106 | session_dir: str, |
| 107 | cache_dir: str, |
| 108 | mode: str = "self-chat", |
| 109 | ) -> bool: |
| 110 | async with _get_bridge_lock(): |
| 111 | return await _ensure_bridge_started( |
| 112 | port=port, |
| 113 | session_dir=session_dir, |
| 114 | cache_dir=cache_dir, |
| 115 | mode=mode, |
| 116 | require_connection=True, |
| 117 | start_label="WhatsApp: starting bridge", |
| 118 | ) |
| 119 | |
| 120 | |
| 121 | async def stop_bridge() -> None: |
| 122 | async with _get_bridge_lock(): |
| 123 | _stop_bridge_process() |
| 124 | |
| 125 | |
| 126 | async def is_bridge_running(port: int) -> bool: |
| 127 | if not _bridge_process or _bridge_process.poll() is not None: |
| 128 | return False |
| 129 | return await _check_health(port) |
| 130 | |
| 131 | |
| 132 | def get_bridge_url(port: int) -> str: |
| 133 | return f"http://127.0.0.1:{port}" |
| 134 | |
| 135 | |
| 136 | async def ensure_bridge_http_up( |
| 137 | port: int, |
| 138 | session_dir: str, |
| 139 | cache_dir: str, |
| 140 | mode: str = "self-chat", |
| 141 | ) -> bool: |
| 142 | """Start bridge if needed and wait for HTTP server only (not WA connection).""" |
| 143 | async with _get_bridge_lock(): |
| 144 | return await _ensure_bridge_started( |
| 145 | port=port, |
| 146 | session_dir=session_dir, |
| 147 | cache_dir=cache_dir, |
| 148 | mode=mode, |
| 149 | require_connection=False, |
| 150 | start_label="WhatsApp: starting bridge for pairing", |
| 151 | ) |
| 152 | |
| 153 | |
| 154 | def is_process_alive() -> bool: |
| 155 | return _bridge_process is not None and _bridge_process.poll() is None |
| 156 | |
| 157 | |
| 158 | def get_running_config() -> dict: |
| 159 | return dict(_bridge_config) |
| 160 | |
| 161 | |
| 162 | # ------------------------------------------------------------------ |
| 163 | # Internal |
| 164 | # ------------------------------------------------------------------ |
| 165 | |
| 166 | def _get_bridge_lock() -> asyncio.Lock: |
| 167 | global _bridge_lock, _bridge_lock_loop |
| 168 | loop = asyncio.get_running_loop() |
| 169 | if _bridge_lock is None or _bridge_lock_loop is not loop: |
| 170 | _bridge_lock = asyncio.Lock() |
| 171 | _bridge_lock_loop = loop |
| 172 | return _bridge_lock |
| 173 | |
| 174 | |
| 175 | async def _ensure_bridge_started( |
| 176 | *, |
| 177 | port: int, |
| 178 | session_dir: str, |
| 179 | cache_dir: str, |
| 180 | mode: str, |
| 181 | require_connection: bool, |
| 182 | start_label: str, |
| 183 | ) -> bool: |
| 184 | global _bridge_process |
| 185 | |
| 186 | if _bridge_process and _bridge_process.poll() is None: |
| 187 | if require_connection: |
| 188 | return True |
| 189 | if await _check_http_up(port): |
| 190 | return True |
| 191 | |
| 192 | PrintStyle.warning("WhatsApp: bridge is running but HTTP is not responding, restarting") |
| 193 | _stop_bridge_process() |
| 194 | |
| 195 | await _ensure_bridge_dependencies() |
| 196 | |
| 197 | attempt = 0 |
| 198 | while attempt < 2: |
| 199 | attempt += 1 |
| 200 | success, output = await _start_bridge_once( |
| 201 | port=port, |
| 202 | session_dir=session_dir, |
| 203 | cache_dir=cache_dir, |
| 204 | mode=mode, |
| 205 | require_connection=require_connection, |
| 206 | start_label=start_label, |
| 207 | ) |
| 208 | if success: |
| 209 | return True |
| 210 | |
| 211 | if attempt == 1 and _looks_like_dependency_failure(output): |
| 212 | PrintStyle.warning( |
| 213 | "WhatsApp: bridge startup looks like a dependency issue, " |
| 214 | "reinstalling dependencies and retrying" |
| 215 | ) |
| 216 | await _ensure_bridge_dependencies(force_reinstall=True) |
| 217 | continue |
| 218 | |
| 219 | return False |
| 220 | |
| 221 | return False |
| 222 | |
| 223 | |
| 224 | async def _start_bridge_once( |
| 225 | *, |
| 226 | port: int, |
| 227 | session_dir: str, |
| 228 | cache_dir: str, |
| 229 | mode: str, |
| 230 | require_connection: bool, |
| 231 | start_label: str, |
| 232 | ) -> tuple[bool, str]: |
| 233 | global _bridge_process |
| 234 | |
| 235 | cmd = [ |
| 236 | "node", BRIDGE_SCRIPT, |
| 237 | "--port", str(port), |
| 238 | "--session", session_dir, |
| 239 | "--cache-dir", cache_dir, |
| 240 | "--mode", mode, |
| 241 | ] |
| 242 | |
| 243 | _kill_port_process(port) |
| 244 | PrintStyle.info(start_label) |
| 245 | _bridge_process = _BridgeProcess(subprocess.Popen( |
| 246 | cmd, |
| 247 | stdout=subprocess.PIPE, |
| 248 | stderr=subprocess.STDOUT, |
| 249 | cwd=BRIDGE_DIR, |
| 250 | ), port) |
| 251 | _start_log_reader(_bridge_process) |
| 252 | _bridge_config.clear() |
| 253 | _bridge_config.update({"port": port, "mode": mode}) |
| 254 | |
| 255 | healthy, output = await _wait_for_bridge_startup( |
| 256 | port=port, |
| 257 | require_connection=require_connection, |
| 258 | ) |
| 259 | if healthy: |
| 260 | return True, output |
| 261 | |
| 262 | if output: |
| 263 | PrintStyle.error(f"WhatsApp: bridge startup failed\n{output}") |
| 264 | return False, output |
| 265 | |
| 266 | |
| 267 | async def _wait_for_bridge_startup(*, port: int, require_connection: bool) -> tuple[bool, str]: |
| 268 | for _ in range(STARTUP_WAIT_ATTEMPTS): |
| 269 | await asyncio.sleep(STARTUP_WAIT_SECONDS) |
| 270 | |
| 271 | process = _bridge_process |
| 272 | if process is None: |
| 273 | return False, "" |
| 274 | |
| 275 | if process.poll() is not None: |
| 276 | output = _summarize_output(process.recent_output()) |
| 277 | PrintStyle.error("WhatsApp: bridge process exited unexpectedly") |
| 278 | _clear_bridge_process() |
| 279 | return False, output |
| 280 | |
| 281 | if require_connection: |
| 282 | if await _check_health(port): |
| 283 | return True, process.recent_output() |
| 284 | else: |
| 285 | if await _check_http_up(port): |
| 286 | return True, process.recent_output() |
| 287 | |
| 288 | if require_connection: |
| 289 | PrintStyle.warning("WhatsApp: bridge started but not yet connected") |
| 290 | process = _bridge_process |
| 291 | return True, process.recent_output() if process else "" |
| 292 | |
| 293 | process = _bridge_process |
| 294 | return False, _summarize_output(process.recent_output()) if process else "" |
| 295 | |
| 296 | |
| 297 | def _looks_like_dependency_failure(output: str) -> bool: |
| 298 | return any(marker in output for marker in DEPENDENCY_FAILURE_MARKERS) |
| 299 | |
| 300 | |
| 301 | async def _check_health(port: int) -> bool: |
| 302 | try: |
| 303 | from plugins._whatsapp_integration.helpers.wa_client import get_health |
| 304 | health = await get_health(get_bridge_url(port)) |
| 305 | return health.get("status") == "connected" |
| 306 | except Exception: |
| 307 | return False |
| 308 | |
| 309 | |
| 310 | async def _check_http_up(port: int) -> bool: |
| 311 | try: |
| 312 | from plugins._whatsapp_integration.helpers.wa_client import get_health |
| 313 | await get_health(get_bridge_url(port)) |
| 314 | return True |
| 315 | except Exception: |
| 316 | return False |
| 317 | |
| 318 | |
| 319 | async def _ensure_bridge_dependencies(force_reinstall: bool = False) -> None: |
| 320 | expected_state = await _build_dependency_state() |
| 321 | |
| 322 | if not force_reinstall: |
| 323 | install_state = _load_dependency_state() |
| 324 | if os.path.isdir(NODE_MODULES_DIR) and await _validate_bridge_dependencies(): |
| 325 | if install_state is None: |
| 326 | _write_dependency_state(expected_state) |
| 327 | return |
| 328 | if install_state == expected_state: |
| 329 | return |
| 330 | |
| 331 | await _reinstall_bridge_dependencies() |
| 332 | |
| 333 | if not await _validate_bridge_dependencies(): |
| 334 | raise RuntimeError("WhatsApp: bridge dependencies failed validation after reinstall") |
| 335 | |
| 336 | _write_dependency_state(await _build_dependency_state()) |
| 337 | |
| 338 | |
| 339 | async def _build_dependency_state() -> dict[str, Any]: |
| 340 | return { |
| 341 | "package_json_hash": _sha256_file(BRIDGE_PACKAGE_JSON), |
| 342 | "package_lock_hash": _sha256_file(BRIDGE_PACKAGE_LOCK) if os.path.isfile(BRIDGE_PACKAGE_LOCK) else "", |
| 343 | "platform": platform.system(), |
| 344 | "arch": platform.machine(), |
| 345 | "node_version": (await _run_subprocess(["node", "--version"], cwd=BRIDGE_DIR)).strip(), |
| 346 | "npm_version": (await _run_subprocess(["npm", "--version"], cwd=BRIDGE_DIR)).strip(), |
| 347 | } |
| 348 | |
| 349 | |
| 350 | async def _validate_bridge_dependencies() -> bool: |
| 351 | dependency_names = _bridge_dependency_names() |
| 352 | if not dependency_names or not os.path.isdir(NODE_MODULES_DIR): |
| 353 | return False |
| 354 | |
| 355 | imports = ", ".join(json.dumps(name) for name in dependency_names) |
| 356 | script = ( |
| 357 | f"for (const name of [{imports}]) {{ await import(name); }}\n" |
| 358 | "process.stdout.write('ok');\n" |
| 359 | ) |
| 360 | |
| 361 | try: |
| 362 | output = await _run_subprocess( |
| 363 | ["node", "--input-type=module", "--eval", script], |
| 364 | cwd=BRIDGE_DIR, |
| 365 | ) |
| 366 | except RuntimeError as e: |
| 367 | PrintStyle.warning(f"WhatsApp: dependency validation failed: {e}") |
| 368 | return False |
| 369 | return output.strip() == "ok" |
| 370 | |
| 371 | |
| 372 | async def _reinstall_bridge_dependencies() -> None: |
| 373 | _ensure_runtime_dir() |
| 374 | |
| 375 | if os.path.isdir(NODE_MODULES_DIR): |
| 376 | PrintStyle.warning("WhatsApp: bridge dependencies missing, outdated, or corrupt; reinstalling") |
| 377 | shutil.rmtree(NODE_MODULES_DIR, ignore_errors=True) |
| 378 | else: |
| 379 | PrintStyle.info("WhatsApp: installing bridge dependencies") |
| 380 | |
| 381 | if os.path.isfile(BRIDGE_INSTALL_STATE): |
| 382 | try: |
| 383 | os.remove(BRIDGE_INSTALL_STATE) |
| 384 | except OSError: |
| 385 | pass |
| 386 | |
| 387 | commands: list[list[str]] = [] |
| 388 | if os.path.isfile(BRIDGE_PACKAGE_LOCK): |
| 389 | commands.append(["npm", "ci", "--omit=dev", "--no-audit", "--no-fund"]) |
| 390 | commands.append(["npm", "install", "--omit=dev", "--no-audit", "--no-fund"]) |
| 391 | env = {"npm_config_cache": BRIDGE_NPM_CACHE} |
| 392 | |
| 393 | last_error: RuntimeError | None = None |
| 394 | for command in commands: |
| 395 | try: |
| 396 | await _run_subprocess(command, cwd=BRIDGE_DIR, env=env) |
| 397 | return |
| 398 | except RuntimeError as e: |
| 399 | last_error = e |
| 400 | PrintStyle.warning(f"WhatsApp: {' '.join(command)} failed: {e}") |
| 401 | |
| 402 | raise RuntimeError(str(last_error) if last_error else "npm install failed") |
| 403 | |
| 404 | |
| 405 | def _bridge_dependency_names() -> list[str]: |
| 406 | with open(BRIDGE_PACKAGE_JSON, "r", encoding="utf-8") as f: |
| 407 | package_json = json.load(f) |
| 408 | dependencies = package_json.get("dependencies") or {} |
| 409 | return sorted(dependencies.keys()) |
| 410 | |
| 411 | |
| 412 | def _load_dependency_state() -> dict[str, Any] | None: |
| 413 | if not os.path.isfile(BRIDGE_INSTALL_STATE): |
| 414 | return None |
| 415 | try: |
| 416 | with open(BRIDGE_INSTALL_STATE, "r", encoding="utf-8") as f: |
| 417 | return json.load(f) |
| 418 | except (OSError, json.JSONDecodeError): |
| 419 | return None |
| 420 | |
| 421 | |
| 422 | def _write_dependency_state(state: dict[str, Any]) -> None: |
| 423 | _ensure_runtime_dir() |
| 424 | with open(BRIDGE_INSTALL_STATE, "w", encoding="utf-8") as f: |
| 425 | json.dump(state, f, indent=2, sort_keys=True) |
| 426 | |
| 427 | |
| 428 | def _sha256_file(path: str) -> str: |
| 429 | digest = hashlib.sha256() |
| 430 | with open(path, "rb") as f: |
| 431 | for chunk in iter(lambda: f.read(65536), b""): |
| 432 | digest.update(chunk) |
| 433 | return digest.hexdigest() |
| 434 | |
| 435 | |
| 436 | def _ensure_runtime_dir() -> None: |
| 437 | os.makedirs(BRIDGE_RUNTIME_DIR, exist_ok=True) |
| 438 | |
| 439 | |
| 440 | async def _run_subprocess( |
| 441 | command: Sequence[str], |
| 442 | *, |
| 443 | cwd: str, |
| 444 | env: dict[str, str] | None = None, |
| 445 | ) -> str: |
| 446 | process_env = os.environ.copy() |
| 447 | if env: |
| 448 | process_env.update(env) |
| 449 | |
| 450 | proc = await asyncio.create_subprocess_exec( |
| 451 | *command, |
| 452 | cwd=cwd, |
| 453 | env=process_env, |
| 454 | stdout=asyncio.subprocess.PIPE, |
| 455 | stderr=asyncio.subprocess.STDOUT, |
| 456 | ) |
| 457 | stdout, _ = await proc.communicate() |
| 458 | output = stdout.decode("utf-8", errors="replace").strip() |
| 459 | if proc.returncode != 0: |
| 460 | raise RuntimeError(output or f"{' '.join(command)} exited with code {proc.returncode}") |
| 461 | return output |
| 462 | |
| 463 | |
| 464 | def _stop_bridge_process() -> None: |
| 465 | global _bridge_process |
| 466 | if not _bridge_process: |
| 467 | return |
| 468 | try: |
| 469 | _bridge_process.terminate() |
| 470 | try: |
| 471 | _bridge_process.wait(timeout=5) |
| 472 | except subprocess.TimeoutExpired: |
| 473 | _bridge_process.kill() |
| 474 | except Exception: |
| 475 | pass |
| 476 | _clear_bridge_process() |
| 477 | PrintStyle.info("WhatsApp: bridge stopped") |
| 478 | |
| 479 | |
| 480 | def _clear_bridge_process() -> None: |
| 481 | global _bridge_process |
| 482 | _bridge_process = None |
| 483 | _bridge_config.clear() |
| 484 | |
| 485 | |
| 486 | def _kill_port_process(port: int) -> None: |
| 487 | """Kill any orphaned process listening on the given TCP port.""" |
| 488 | try: |
| 489 | system = platform.system() |
| 490 | if system == "Windows": |
| 491 | result = subprocess.run( |
| 492 | ["netstat", "-ano", "-p", "TCP"], |
| 493 | capture_output=True, text=True, timeout=5, |
| 494 | ) |
| 495 | for line in result.stdout.splitlines(): |
| 496 | parts = line.split() |
| 497 | if len(parts) >= 5 and parts[3] == "LISTENING": |
| 498 | if parts[1].endswith(f":{port}"): |
| 499 | try: |
| 500 | subprocess.run( |
| 501 | ["taskkill", "/PID", parts[4], "/F"], |
| 502 | capture_output=True, timeout=5, |
| 503 | ) |
| 504 | except subprocess.SubprocessError: |
| 505 | pass |
| 506 | elif system == "Darwin": |
| 507 | result = subprocess.run( |
| 508 | ["lsof", "-ti", f"tcp:{port}"], |
| 509 | capture_output=True, text=True, timeout=5, |
| 510 | ) |
| 511 | for pid_str in result.stdout.strip().splitlines(): |
| 512 | try: |
| 513 | os.kill(int(pid_str.strip()), 9) |
| 514 | except (ValueError, OSError): |
| 515 | pass |
| 516 | else: |
| 517 | result = subprocess.run( |
| 518 | ["fuser", f"{port}/tcp"], |
| 519 | capture_output=True, timeout=5, |
| 520 | ) |
| 521 | if result.returncode == 0: |
| 522 | subprocess.run( |
| 523 | ["fuser", "-k", f"{port}/tcp"], |
| 524 | capture_output=True, timeout=5, |
| 525 | ) |
| 526 | except Exception: |
| 527 | pass |
| 528 | |
| 529 | |
| 530 | def _start_log_reader(process: _BridgeProcess) -> None: |
| 531 | def _reader() -> None: |
| 532 | assert process.stdout |
| 533 | for line in iter(process.stdout.readline, b""): |
| 534 | text = line.decode("utf-8", errors="replace").rstrip() |
| 535 | if text: |
| 536 | process.remember_output(text) |
| 537 | PrintStyle.standard(f"WhatsApp bridge: {text}") |
| 538 | process.stdout.close() |
| 539 | |
| 540 | thread = threading.Thread(target=_reader, daemon=True) |
| 541 | thread.start() |
| 542 | |
| 543 | |
| 544 | def _summarize_output(output: str, max_lines: int = 12) -> str: |
| 545 | if not output: |
| 546 | return "" |
| 547 | lines = [line for line in output.splitlines() if line.strip()] |
| 548 | return "\n".join(lines[-max_lines:]) |