From ab37f15b2746f38f41195d03598115102e5c54f4 Mon Sep 17 00:00:00 2001 From: Amit Bhakt Date: Wed, 21 Oct 2026 23:51:56 +0530 Subject: [PATCH] feat: Add native WhatsApp Cloud API serving with SQLite session persistence - Add Agent.serve_whatsapp() and Server.serve_whatsapp() - Implement SQLiteSessionStore with WAL mode, session timeout, and message idempotency - Implement WhatsAppServer with HMAC-SHA256 signature verification and background task processing - Implement MetaWhatsAppClient with message chunking and read receipts - Implement MetaDiscoveryClient for automated Phone Number ID discovery and 3-tier webhook sync (adapted from WAVoice) - Add 17 unit and integration tests covering security, idempotency, session isolation, and auto-discovery --- .gitignore | 5 + src/sdialog/agents.py | 59 +++ src/sdialog/server.py | 63 ++++ src/sdialog/whatsapp/__init__.py | 25 ++ src/sdialog/whatsapp/client.py | 121 ++++++ src/sdialog/whatsapp/config.py | 64 ++++ src/sdialog/whatsapp/discovery.py | 181 +++++++++ src/sdialog/whatsapp/server.py | 331 +++++++++++++++++ src/sdialog/whatsapp/session_store.py | 214 +++++++++++ tests/test_whatsapp.py | 514 ++++++++++++++++++++++++++ 10 files changed, 1577 insertions(+) create mode 100644 src/sdialog/whatsapp/__init__.py create mode 100644 src/sdialog/whatsapp/client.py create mode 100644 src/sdialog/whatsapp/config.py create mode 100644 src/sdialog/whatsapp/discovery.py create mode 100644 src/sdialog/whatsapp/server.py create mode 100644 src/sdialog/whatsapp/session_store.py create mode 100644 tests/test_whatsapp.py diff --git a/.gitignore b/.gitignore index 79120d5..61f9e2a 100644 --- a/.gitignore +++ b/.gitignore @@ -48,3 +48,8 @@ tutorials/01_audio/dscaper_data_impulse_response/* *audio_dialog.json ./src/sdialog/sdialog tutorials/01_audio/*.png + +# SQLite +*.db +*.db-wal +*.db-shm diff --git a/src/sdialog/agents.py b/src/sdialog/agents.py index 92abbbe..954e6e4 100644 --- a/src/sdialog/agents.py +++ b/src/sdialog/agents.py @@ -670,6 +670,65 @@ def serve(self, return Server.serve(agents=self, host=host, port=port, stateless=stateless, debug=debug, log_level=log_level) + def serve_whatsapp(self, + port: int = 1234, + host: str = "0.0.0.0", + phone_number_id: Optional[str] = None, + access_token: Optional[str] = None, + verify_token: Optional[str] = None, + app_secret: Optional[str] = None, + db_path: str = "whatsapp_sessions.db", + webhook_path: str = "/webhook", + session_timeout: int = 1800, + tunnel: bool = False, + auto_setup: bool = True, + log_level: str = "info"): + """ + Starts a Meta WhatsApp Cloud API webhook server to interact with the agent. + + :param port: Port number to bind the server to. + :type port: int + :param host: Host address to bind the server to. + :type host: str + :param phone_number_id: Meta WhatsApp Phone Number ID (or WHATSAPP_PHONE_NUMBER_ID env var). + :type phone_number_id: Optional[str] + :param access_token: Meta Graph API Access Token (or WHATSAPP_ACCESS_TOKEN env var). + :type access_token: Optional[str] + :param verify_token: Webhook verification token (or WHATSAPP_VERIFY_TOKEN env var). + :type verify_token: Optional[str] + :param app_secret: Meta App Secret for HMAC validation (or WHATSAPP_APP_SECRET env var). + :type app_secret: Optional[str] + :param db_path: Path to the SQLite session database. + :type db_path: str + :param webhook_path: Webhook URL path. + :type webhook_path: str + :param session_timeout: Seconds of inactivity before a user session resets. + :type session_timeout: int + :param tunnel: If True, attempts to spawn a public ngrok tunnel. + :type tunnel: bool + :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :type auto_setup: bool + :param log_level: Logging level for the server. + :type log_level: str + """ + from .whatsapp import WhatsAppServer, WhatsAppConfig + + config = WhatsAppConfig( + port=port, + host=host, + phone_number_id=phone_number_id, + access_token=access_token, + verify_token=verify_token, + app_secret=app_secret, + db_path=db_path, + webhook_path=webhook_path, + session_timeout=session_timeout, + tunnel=tunnel, + auto_setup=auto_setup, + log_level=log_level + ) + return WhatsAppServer.serve(agent=self, config=config) + def response_lookahead(self, message: str = None): """ Generates a response without updating the agent's memory. diff --git a/src/sdialog/server.py b/src/sdialog/server.py index 5119000..bf0420e 100644 --- a/src/sdialog/server.py +++ b/src/sdialog/server.py @@ -318,6 +318,69 @@ async def serve_async(cls, server = uvicorn.Server(config) await server.serve() + @classmethod + def serve_whatsapp(cls, + agent: Agent, + port: int = 1234, + host: str = "0.0.0.0", + phone_number_id: Optional[str] = None, + access_token: Optional[str] = None, + verify_token: Optional[str] = None, + app_secret: Optional[str] = None, + db_path: str = "whatsapp_sessions.db", + webhook_path: str = "/webhook", + session_timeout: int = 1800, + tunnel: bool = False, + auto_setup: bool = True, + log_level: str = "info") -> None: + """ + Serve an SDialog agent via Meta WhatsApp Cloud API webhook. + + :param agent: The SDialog agent to serve. + :type agent: Agent + :param port: Port number to listen on. + :type port: int + :param host: Host address to bind to. + :type host: str + :param phone_number_id: Meta WhatsApp Phone Number ID. + :type phone_number_id: Optional[str] + :param access_token: Meta Graph API Access Token. + :type access_token: Optional[str] + :param verify_token: Webhook verification token. + :type verify_token: Optional[str] + :param app_secret: Meta App Secret for HMAC validation. + :type app_secret: Optional[str] + :param db_path: Path to the SQLite session database. + :type db_path: str + :param webhook_path: Webhook URL path. + :type webhook_path: str + :param session_timeout: Inactivity seconds before resetting session. + :type session_timeout: int + :param tunnel: If True, attempts to spawn a public ngrok tunnel. + :type tunnel: bool + :param auto_setup: If True, auto-discovers phone ID and auto-syncs Meta webhooks. + :type auto_setup: bool + :param log_level: Logging level for the server. + :type log_level: str + """ + from .whatsapp import WhatsAppServer, WhatsAppConfig + + config = WhatsAppConfig( + port=port, + host=host, + phone_number_id=phone_number_id, + access_token=access_token, + verify_token=verify_token, + app_secret=app_secret, + db_path=db_path, + webhook_path=webhook_path, + session_timeout=session_timeout, + tunnel=tunnel, + auto_setup=auto_setup, + log_level=log_level + ) + return WhatsAppServer.serve(agent=agent, config=config) + @classmethod def serve_in_thread(cls, agents: Union[Agent, List[Agent]], diff --git a/src/sdialog/whatsapp/__init__.py b/src/sdialog/whatsapp/__init__.py new file mode 100644 index 0000000..6cc40ca --- /dev/null +++ b/src/sdialog/whatsapp/__init__.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +WhatsApp integration for SDialog agents. + +This package provides native WhatsApp serving capabilities for SDialog agents, +enabling two-way text conversations over the Meta WhatsApp Cloud API with SQLite +persistence and per-user session isolation. +""" + +from .config import WhatsAppConfig +from .server import WhatsAppServer +from .session_store import SQLiteSessionStore +from .client import MetaWhatsAppClient +from .discovery import MetaDiscoveryClient + +__all__ = [ + "WhatsAppConfig", + "WhatsAppServer", + "SQLiteSessionStore", + "MetaWhatsAppClient", + "MetaDiscoveryClient", +] diff --git a/src/sdialog/whatsapp/client.py b/src/sdialog/whatsapp/client.py new file mode 100644 index 0000000..b4393af --- /dev/null +++ b/src/sdialog/whatsapp/client.py @@ -0,0 +1,121 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Meta WhatsApp Cloud API client. +""" + +import logging +from typing import List +import httpx + +from .config import WhatsAppConfig + +logger = logging.getLogger(__name__) + +# WhatsApp limits single message text body to 4096 characters +MAX_WHATSAPP_TEXT_LENGTH = 4096 + + +class MetaWhatsAppClient: + """Async client for interacting with the Meta WhatsApp Cloud API.""" + + def __init__(self, config: WhatsAppConfig): + self.config = config + self.base_url = f"https://graph.facebook.com/{config.api_version}/{config.phone_number_id}" + self.headers = { + "Authorization": f"Bearer {config.access_token}", + "Content-Type": "application/json", + } + + async def send_text_message(self, to: str, text: str) -> bool: + """ + Send a text message to a WhatsApp user. + + If the text exceeds WhatsApp's maximum length (4096 characters), + it is split and sent in chunks. + """ + if not text: + return False + + chunks = self._chunk_text(text, MAX_WHATSAPP_TEXT_LENGTH) + success = True + + async with httpx.AsyncClient(timeout=10.0) as client: + for chunk in chunks: + payload = { + "messaging_product": "whatsapp", + "recipient_type": "individual", + "to": to, + "type": "text", + "text": {"preview_url": False, "body": chunk}, + } + + try: + response = await client.post( + f"{self.base_url}/messages", + json=payload, + headers=self.headers, + ) + + if response.status_code in (200, 201): + logger.debug(f"[WhatsApp] Sent message to {to}") + else: + logger.error( + f"[WhatsApp] Failed to send message to {to}. " + f"HTTP {response.status_code}: {response.text}" + ) + success = False + except Exception as e: + logger.error(f"[WhatsApp] Exception while sending message to {to}: {e}") + success = False + + return success + + async def mark_as_read(self, message_id: str) -> bool: + """Mark an incoming WhatsApp message as read (blue double check).""" + if not message_id: + return False + + payload = { + "messaging_product": "whatsapp", + "status": "read", + "message_id": message_id, + } + + async with httpx.AsyncClient(timeout=5.0) as client: + try: + response = await client.post( + f"{self.base_url}/messages", + json=payload, + headers=self.headers, + ) + return response.status_code == 200 + except Exception as e: + logger.debug(f"[WhatsApp] Could not mark message {message_id} as read: {e}") + return False + + @staticmethod + def _chunk_text(text: str, max_length: int) -> List[str]: + """Split long messages into WhatsApp-safe chunks without splitting words where possible.""" + if len(text) <= max_length: + return [text] + + chunks = [] + while text: + if len(text) <= max_length: + chunks.append(text) + break + + # Find last whitespace within max_length + split_idx = text.rfind("\n", 0, max_length) + if split_idx == -1: + split_idx = text.rfind(" ", 0, max_length) + if split_idx == -1: + split_idx = max_length + + chunks.append(text[:split_idx].strip()) + text = text[split_idx:].lstrip() + + return chunks diff --git a/src/sdialog/whatsapp/config.py b/src/sdialog/whatsapp/config.py new file mode 100644 index 0000000..9843fc0 --- /dev/null +++ b/src/sdialog/whatsapp/config.py @@ -0,0 +1,64 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Configuration models for SDialog WhatsApp server. +""" + +import os +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass +class WhatsAppConfig: + """ + Configuration parameters for serving an SDialog agent on WhatsApp. + + Parameters will fall back to standard environment variables if not provided: + - ``WHATSAPP_PHONE_NUMBER_ID`` + - ``WHATSAPP_ACCESS_TOKEN`` + - ``WHATSAPP_VERIFY_TOKEN`` + - ``WHATSAPP_APP_SECRET`` + """ + + port: int = 1234 + host: str = "0.0.0.0" + phone_number_id: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_PHONE_NUMBER_ID") + ) + access_token: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_ACCESS_TOKEN") or os.environ.get("WHATSAPP_TOKEN") + ) + verify_token: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_VERIFY_TOKEN") + or os.environ.get("WHATSAPP_WEBHOOK_VERIFY_TOKEN", "sdialog_whatsapp_secret") + ) + app_secret: Optional[str] = field( + default_factory=lambda: os.environ.get("WHATSAPP_APP_SECRET") + ) + api_version: str = "v21.0" + db_path: str = "whatsapp_sessions.db" + webhook_path: str = "/webhook" + session_timeout: int = 1800 # 30 minutes idle timeout + tunnel: bool = False + auto_setup: bool = True # Automatically discover IDs and configure Meta webhooks + max_history_turns: int = 40 # Context window sliding budget + enforce_signature: bool = True + mark_as_read: bool = True + log_level: str = "info" + + def validate(self, dry_run: bool = False) -> None: + """Validate that required credentials are set when not in dry_run mode.""" + if not dry_run: + if not self.phone_number_id: + raise ValueError( + "WhatsApp phone_number_id is required. Pass it to serve_whatsapp() " + "or set the WHATSAPP_PHONE_NUMBER_ID environment variable." + ) + if not self.access_token: + raise ValueError( + "WhatsApp access_token is required. Pass it to serve_whatsapp() " + "or set the WHATSAPP_ACCESS_TOKEN environment variable." + ) diff --git a/src/sdialog/whatsapp/discovery.py b/src/sdialog/whatsapp/discovery.py new file mode 100644 index 0000000..2a82dff --- /dev/null +++ b/src/sdialog/whatsapp/discovery.py @@ -0,0 +1,181 @@ +# SPDX-FileCopyrightText: Copyright © 2026 Amit Bhakt +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-License-Identifier: MIT +# +# Portions of this module are adapted from WAVoice: +# "WAVoice: WebRTC Gateway and DevOps Automation Engine for WhatsApp Voice AI" +# Repository: https://github.com/amitbhakt/wavoice +# DOI: 10.5281/zenodo.23537489 +# Author: Amit Bhakt +# Licensed under the MIT License. +""" +Meta WhatsApp Cloud API auto-discovery and multi-tier webhook configuration. + +This module automates the onboarding process for WhatsApp Cloud API agents: +1. Token inspection (extracts App ID, WABA ID, permissions) +2. Automatic discovery of Phone Number ID and WABA ID +3. 3-Tier Webhook synchronization (App-level, WABA-level, and Phone-number-level) +""" + +import logging +from typing import Optional, Dict, Any, List +import httpx + +logger = logging.getLogger(__name__) + +GRAPH_API_VERSION = "v21.0" +GRAPH_BASE_URL = f"https://graph.facebook.com/{GRAPH_API_VERSION}" + + +class MetaDiscoveryClient: + """ + Client for automating Meta WhatsApp Cloud API credentials discovery and webhook setup. + + Adapted from WAVoice (MIT License, https://github.com/amitbhakt/wavoice). + """ + + def __init__(self, access_token: str, app_secret: Optional[str] = None): + self.access_token = access_token + self.app_secret = app_secret + + @property + def _headers(self) -> Dict[str, str]: + return { + "Authorization": f"Bearer {self.access_token}", + "Content-Type": "application/json", + } + + async def inspect_token(self) -> Dict[str, Any]: + """ + Inspect the provided User/System Access Token via debug_token endpoint. + + Returns metadata including app_id, scopes, user_id, and granular target IDs (WABAs). + """ + url = f"{GRAPH_BASE_URL}/debug_token" + params = {"input_token": self.access_token, "access_token": self.access_token} + + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(url, params=params) + data = resp.json() + if resp.status_code != 200: + error_msg = data.get("error", {}).get("message", "Token inspection failed") + raise ValueError(f"Invalid Meta Access Token: {error_msg}") + return data.get("data", {}) + + async def discover_waba_id(self) -> Optional[str]: + """Auto-discover WhatsApp Business Account (WABA) ID from token permissions.""" + try: + token_data = await self.inspect_token() + granular_scopes = token_data.get("granular_scopes", []) + for scope_info in granular_scopes: + if scope_info.get("scope") in ("whatsapp_business_management", "whatsapp_business_messaging"): + target_ids = scope_info.get("target_ids", []) + if target_ids: + return str(target_ids[0]) + except Exception as e: + logger.debug(f"[WhatsApp] Could not auto-discover WABA ID: {e}") + return None + + async def list_phone_numbers(self, waba_id: str) -> List[Dict[str, Any]]: + """List all phone numbers under a WhatsApp Business Account.""" + url = f"{GRAPH_BASE_URL}/{waba_id}/phone_numbers" + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(url, headers=self._headers) + if resp.status_code == 200: + return resp.json().get("data", []) + return [] + + async def discover_phone_number_id(self, waba_id: Optional[str] = None) -> Optional[str]: + """Auto-discover the primary WhatsApp Phone Number ID.""" + target_waba = waba_id or await self.discover_waba_id() + if not target_waba: + return None + + numbers = await self.list_phone_numbers(target_waba) + if numbers: + phone_id = str(numbers[0].get("id")) + display_num = numbers[0].get("display_phone_number", "unknown") + logger.info(f"[WhatsApp] Auto-discovered Phone Number ID: {phone_id} ({display_num})") + return phone_id + return None + + async def configure_webhooks( + self, + callback_url: str, + verify_token: str, + phone_number_id: Optional[str] = None, + waba_id: Optional[str] = None, + ) -> Dict[str, bool]: + """ + Programmatically synchronize all 3 tiers of Meta WhatsApp webhooks: + 1. App-level subscription (POST /{app_id}/subscriptions) + 2. WABA-level subscription (POST /{waba_id}/subscribed_apps) + 3. Phone-number-level override (POST /{phone_number_id}) + """ + results = {"app_level": False, "waba_level": False, "phone_level": False} + + token_info = await self.inspect_token() + app_id = token_info.get("app_id") + target_waba = waba_id or await self.discover_waba_id() + + async with httpx.AsyncClient(timeout=15.0) as client: + # Tier 1: App-Level Webhook Subscription (requires App Access Token) + if app_id and self.app_secret: + try: + app_access_token = f"{app_id}|{self.app_secret}" + app_sub_url = f"{GRAPH_BASE_URL}/{app_id}/subscriptions" + payload = { + "object": "whatsapp_business_account", + "callback_url": callback_url, + "verify_token": verify_token, + "fields": "messages", + "include_values": "true", + "access_token": app_access_token, + } + resp = await client.post(app_sub_url, data=payload) + if resp.status_code == 200 and resp.json().get("success"): + results["app_level"] = True + logger.info("[WhatsApp] Configured App-level webhook subscription") + else: + logger.warning(f"[WhatsApp] App-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set App-level webhook: {e}") + + # Tier 2: WABA-Level Subscribed Apps + if target_waba: + try: + waba_sub_url = f"{GRAPH_BASE_URL}/{target_waba}/subscribed_apps" + waba_payload = { + "override_callback_uri": callback_url, + "verify_token": verify_token, + } + resp = await client.post(waba_sub_url, headers=self._headers, json=waba_payload) + if resp.status_code == 200 and resp.json().get("success"): + results["waba_level"] = True + logger.info(f"[WhatsApp] Configured WABA-level webhook override on {target_waba}") + else: + logger.warning(f"[WhatsApp] WABA-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set WABA-level webhook: {e}") + + # Tier 3: Phone-Number-Level Webhook Override + target_phone = phone_number_id or await self.discover_phone_number_id(target_waba) + if target_phone: + try: + phone_url = f"{GRAPH_BASE_URL}/{target_phone}" + phone_payload = { + "webhook_configuration": { + "override_callback_uri": callback_url, + "verify_token": verify_token, + } + } + resp = await client.post(phone_url, headers=self._headers, json=phone_payload) + if resp.status_code == 200 and resp.json().get("success"): + results["phone_level"] = True + logger.info(f"[WhatsApp] Configured Phone-level webhook override on {target_phone}") + else: + logger.warning(f"[WhatsApp] Phone-level webhook config returned: {resp.text}") + except Exception as e: + logger.warning(f"[WhatsApp] Could not set Phone-level webhook: {e}") + + return results diff --git a/src/sdialog/whatsapp/server.py b/src/sdialog/whatsapp/server.py new file mode 100644 index 0000000..a6fa958 --- /dev/null +++ b/src/sdialog/whatsapp/server.py @@ -0,0 +1,331 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +FastAPI webhook server for SDialog WhatsApp agent serving. +""" + +import hmac +import hashlib +import asyncio +import logging +import threading +from typing import Optional, Dict, Any + +from fastapi import FastAPI, Request, Response, BackgroundTasks, HTTPException, Query +from fastapi.responses import JSONResponse +import uvicorn + +from .config import WhatsAppConfig +from .client import MetaWhatsAppClient +from .session_store import ( + SQLiteSessionStore, + restore_agent_state, + dump_agent_state, +) + +logger = logging.getLogger(__name__) + + +class WhatsAppServer: + """ + FastAPI server exposing Meta WhatsApp Cloud API webhook endpoints for SDialog agents. + """ + + _app: Optional[FastAPI] = None + _config: Optional[WhatsAppConfig] = None + _agent: Any = None + _store: Optional[SQLiteSessionStore] = None + _client: Optional[MetaWhatsAppClient] = None + _user_locks: Dict[str, asyncio.Lock] = {} + + @classmethod + def get_user_lock(cls, wa_id: str) -> asyncio.Lock: + """Get or create an async lock for a specific WhatsApp sender ID.""" + if wa_id not in cls._user_locks: + cls._user_locks[wa_id] = asyncio.Lock() + return cls._user_locks[wa_id] + + @classmethod + def create_app(cls, agent: Any, config: WhatsAppConfig) -> FastAPI: + """Create and configure the FastAPI WhatsApp webhook application.""" + cls._agent = agent + cls._config = config + cls._store = SQLiteSessionStore(db_path=config.db_path) + cls._client = MetaWhatsAppClient(config) + cls._user_locks.clear() + + app = FastAPI( + title="SDialog WhatsApp Agent Server", + description="Meta WhatsApp Cloud API Webhook Server for SDialog Agents", + version="1.0.0", + ) + + @app.get(config.webhook_path) + async def verify_webhook( + hub_mode: str = Query(None, alias="hub.mode"), + hub_verify_token: str = Query(None, alias="hub.verify_token"), + hub_challenge: str = Query(None, alias="hub.challenge"), + ): + """Handle Meta WhatsApp webhook verification challenge.""" + if hub_mode == "subscribe" and hub_verify_token == config.verify_token: + logger.info("[WhatsApp] Webhook verification challenge accepted") + return Response(content=hub_challenge, media_type="text/plain") + logger.warning("[WhatsApp] Webhook verification failed: token mismatch") + return Response(content="Verification failed", status_code=403) + + @app.post(config.webhook_path) + async def receive_webhook(request: Request, background_tasks: BackgroundTasks): + """ + Receive incoming WhatsApp webhook events. + + Validates HMAC signature, checks idempotency, and queues background processing. + Returns 200 OK immediately to satisfy Meta's 3-second SLA. + """ + raw_body = await request.body() + + # 1. HMAC Signature Verification + if config.app_secret and config.enforce_signature: + sig_header = request.headers.get("X-Hub-Signature-256", "") + if not cls._verify_signature(raw_body, sig_header, config.app_secret): + logger.warning("[WhatsApp] Rejected request with invalid HMAC signature") + raise HTTPException(status_code=401, detail="Invalid signature") + + # 2. Parse payload + try: + payload = await request.json() + except Exception: + return JSONResponse({"status": "invalid_json"}, status_code=400) + + # 3. Guard: Extract messages from Meta payload structure + messages, sender_name = cls._extract_messages(payload) + if not messages: + # Normal delivery status update (sent, delivered, read) - acknowledge and ignore + return {"status": "ok"} + + for msg in messages: + wamid = msg.get("id") + sender_id = msg.get("from") + msg_type = msg.get("type") + + if not sender_id or not wamid: + continue + + # 4. Idempotency check: drop duplicate webhooks from Meta retries + if cls._store.is_message_processed(wamid): + logger.info(f"[WhatsApp] Duplicate message {wamid} ignored") + continue + + cls._store.mark_message_processed(wamid) + + # 5. Route by message type + if msg_type == "text": + text_body = msg.get("text", {}).get("body", "").strip() + if text_body: + background_tasks.add_task( + cls._handle_user_turn, + sender_id=sender_id, + text=text_body, + wamid=wamid, + ) + else: + # Non-text media (audio, image, document, location, etc.) + background_tasks.add_task( + cls._handle_unsupported_media, + sender_id=sender_id, + msg_type=msg_type, + ) + + return {"status": "ok"} + + @app.get("/health") + async def health(): + """Health check endpoint.""" + return {"status": "healthy", "service": "sdialog-whatsapp"} + + cls._app = app + return app + + @classmethod + async def _handle_user_turn(cls, sender_id: str, text: str, wamid: str) -> None: + """Process incoming user utterance under a per-user lock and reply via WhatsApp.""" + user_lock = cls.get_user_lock(sender_id) + + async with user_lock: + try: + # Mark message as read (blue checkmark) if configured + if cls._config.mark_as_read and cls._client: + await cls._client.mark_as_read(wamid) + + # Check for explicit session reset commands + if text.lower() in ("/reset", "/clear", "/restart"): + cls._store.reset_session(sender_id) + await cls._client.send_text_message( + to=sender_id, + text="Your conversation has been reset. How can I help you?", + ) + return + + # Load user state from SQLite + state = cls._store.load_session( + sender_id, timeout_seconds=cls._config.session_timeout + ) + + # Reset agent memory and restore user session state + cls._agent.reset() + restore_agent_state(cls._agent, state) + + # Run inference in a threadpool so synchronous LLMs don't block the event loop + logger.debug(f"[WhatsApp] Processing turn for user {sender_id}: '{text[:30]}...'") + response = await asyncio.to_thread(cls._agent, text) + + if not response: + logger.warning(f"[WhatsApp] Agent returned empty response for {sender_id}") + return + + # Save updated user state to SQLite + new_state = dump_agent_state( + cls._agent, max_history_turns=cls._config.max_history_turns + ) + cls._store.save_session(sender_id, new_state) + + # Dispatch reply via Meta WhatsApp Cloud API + await cls._client.send_text_message(to=sender_id, text=response) + + except Exception as e: + logger.error(f"[WhatsApp] Error processing message for {sender_id}: {e}", exc_info=True) + try: + await cls._client.send_text_message( + to=sender_id, + text="I encountered a problem processing your request. Please try again in a moment.", + ) + except Exception: + pass + + @classmethod + async def _handle_unsupported_media(cls, sender_id: str, msg_type: str) -> None: + """Notify user that non-text media messages are currently unsupported.""" + logger.info(f"[WhatsApp] Received unsupported message type '{msg_type}' from {sender_id}") + if cls._client: + await cls._client.send_text_message( + to=sender_id, + text="I currently only support text messages. Please send your query as text.", + ) + + @staticmethod + def _verify_signature(raw_body: bytes, sig_header: str, app_secret: str) -> bool: + """Verify the X-Hub-Signature-256 header using HMAC-SHA256 in constant time.""" + if not sig_header or not sig_header.startswith("sha256="): + return False + expected_hash = sig_header[7:] + computed_hash = hmac.new( + app_secret.encode("utf-8"), + msg=raw_body, + digestmod=hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(computed_hash, expected_hash) + + @staticmethod + def _extract_messages(payload: Dict[str, Any]): + """Safely extract message objects and contact details from Meta webhook payload.""" + messages = [] + sender_name = None + try: + entry = payload.get("entry", [])[0] + change = entry.get("changes", [])[0].get("value", {}) + messages = change.get("messages", []) + contacts = change.get("contacts", []) + if contacts: + sender_name = contacts[0].get("profile", {}).get("name") + except (IndexError, KeyError, AttributeError): + pass + return messages, sender_name + + @classmethod + def serve(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> None: + """Start the WhatsApp webhook server with optional auto-discovery and webhook sync.""" + config = config or WhatsAppConfig() + + # 1. Auto-discover Phone Number ID if missing but access token is present + if not config.phone_number_id and config.access_token: + from .discovery import MetaDiscoveryClient + discovery = MetaDiscoveryClient(config.access_token, config.app_secret) + try: + config.phone_number_id = asyncio.run(discovery.discover_phone_number_id()) + except Exception as e: + logger.debug(f"[WhatsApp] Auto-discovery skipped or failed: {e}") + + config.validate(dry_run=False) + + # 2. Optional tunnel setup for local testing + public_url = None + if config.tunnel: + public_url = cls._start_tunnel(config) + + # 3. Auto-configure Meta webhooks if requested and tunnel/public URL is active + if config.auto_setup and public_url and config.app_secret: + from .discovery import MetaDiscoveryClient + discovery = MetaDiscoveryClient(config.access_token, config.app_secret) + callback_url = f"{public_url}{config.webhook_path}" + logger.info(f"[WhatsApp] Auto-configuring Meta webhooks to {callback_url}...") + try: + res = asyncio.run(discovery.configure_webhooks( + callback_url=callback_url, + verify_token=config.verify_token, + phone_number_id=config.phone_number_id + )) + logger.info(f"[WhatsApp] Auto-configuration results: {res}") + except Exception as e: + logger.warning(f"[WhatsApp] Auto-configuration encountered an error: {e}") + + app = cls.create_app(agent, config) + + logger.info(f"[WhatsApp] Starting server on {config.host}:{config.port}") + logger.info(f"[WhatsApp] Webhook endpoint: {config.webhook_path}") + logger.info(f"[WhatsApp] Verify Token: {config.verify_token}") + + uvicorn.run(app, host=config.host, port=config.port, log_level=config.log_level) + + @classmethod + def serve_in_thread(cls, agent: Any, config: Optional[WhatsAppConfig] = None) -> threading.Thread: + """Start the WhatsApp webhook server in a background daemon thread (useful for Jupyter / tests).""" + config = config or WhatsAppConfig() + app = cls.create_app(agent, config) + + def run_server(): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + uv_config = uvicorn.Config( + app, + host=config.host, + port=config.port, + log_level=config.log_level, + loop="asyncio", + ) + server = uvicorn.Server(uv_config) + loop.run_until_complete(server.serve()) + + thread = threading.Thread(target=run_server, daemon=True) + thread.start() + logger.info(f"[WhatsApp] Threaded server started on {config.host}:{config.port}") + return thread + + @staticmethod + def _start_tunnel(config: WhatsAppConfig) -> Optional[str]: + """Attempt to spawn an ngrok tunnel if pyngrok is installed. Returns public URL.""" + try: + from pyngrok import ngrok + public_url = ngrok.connect(config.port, bind_tls=True).public_url + logger.info("=" * 60) + logger.info(f"[WhatsApp] Public Tunnel URL: {public_url}{config.webhook_path}") + logger.info(f"[WhatsApp] Meta Callback URL: {public_url}{config.webhook_path}") + logger.info(f"[WhatsApp] Meta Verify Token: {config.verify_token}") + logger.info("=" * 60) + return public_url + except ImportError: + logger.warning( + "[WhatsApp] tunnel=True requested, but 'pyngrok' is not installed. " + "Install it with `pip install pyngrok` or use a tool like ngrok/cloudflared manually." + ) + return None diff --git a/src/sdialog/whatsapp/session_store.py b/src/sdialog/whatsapp/session_store.py new file mode 100644 index 0000000..948592a --- /dev/null +++ b/src/sdialog/whatsapp/session_store.py @@ -0,0 +1,214 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +SQLite-backed multi-user session and idempotency store for WhatsApp. +""" + +import json +import logging +import sqlite3 +from datetime import datetime +from typing import Optional, Any, Dict, List + +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage + +logger = logging.getLogger(__name__) + +_TYPE_TO_MSG_CLASS = { + "HumanMessage": HumanMessage, + "AIMessage": AIMessage, + "SystemMessage": SystemMessage, +} + + +class SQLiteSessionStore: + """ + Thread-safe SQLite store for WhatsApp user sessions and webhook idempotency. + + Features: + - WAL mode for concurrent non-blocking reads/writes. + - Idempotency tracking to reject duplicate Meta webhook deliveries. + - Automatic session expiration after inactivity timeout. + """ + + def __init__(self, db_path: str = "whatsapp_sessions.db"): + self.db_path = db_path + self._init_db() + + def _get_connection(self) -> sqlite3.Connection: + """Create a new SQLite connection with WAL mode and busy timeout configured.""" + conn = sqlite3.connect(self.db_path, timeout=10.0) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode = WAL;") + conn.execute("PRAGMA busy_timeout = 5000;") + conn.execute("PRAGMA synchronous = NORMAL;") + return conn + + def _init_db(self) -> None: + """Initialize database tables and indexes.""" + with self._get_connection() as conn: + conn.executescript(""" + CREATE TABLE IF NOT EXISTS whatsapp_sessions ( + wa_id TEXT PRIMARY KEY, + state TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_active TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE TABLE IF NOT EXISTS processed_messages ( + wamid TEXT PRIMARY KEY, + received_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_processed_received + ON processed_messages(received_at); + """) + + def load_session(self, wa_id: str, timeout_seconds: int = 1800) -> Optional[Dict[str, Any]]: + """ + Load session state for a user. + + Returns None if session does not exist or has exceeded timeout_seconds of inactivity. + """ + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT state, last_active FROM whatsapp_sessions WHERE wa_id = ?", + (wa_id,) + ) + row = cursor.fetchone() + if not row: + return None + + state_json, last_active_str = row["state"], row["last_active"] + + # Parse last_active timestamp + try: + # Handle SQLite timestamp formats + if "T" in last_active_str: + last_active = datetime.fromisoformat(last_active_str) + else: + last_active = datetime.strptime(last_active_str, "%Y-%m-%d %H:%M:%S") + except Exception: + last_active = datetime.now() + + # Check timeout + elapsed = (datetime.now() - last_active).total_seconds() + if elapsed > timeout_seconds: + logger.info(f"[WhatsApp] Session for user {wa_id} expired ({elapsed:.0f}s > {timeout_seconds}s)") + self.reset_session(wa_id) + return None + + try: + return json.loads(state_json) + except json.JSONDecodeError: + logger.warning(f"[WhatsApp] Corrupted session JSON for {wa_id}, resetting") + self.reset_session(wa_id) + return None + + def save_session(self, wa_id: str, state: Dict[str, Any]) -> None: + """Save or update the session state for a user with the current timestamp.""" + state_json = json.dumps(state) + now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO whatsapp_sessions (wa_id, state, last_active) + VALUES (?, ?, ?) + ON CONFLICT(wa_id) DO UPDATE SET + state = excluded.state, + last_active = excluded.last_active; + """, + (wa_id, state_json, now_str) + ) + + def reset_session(self, wa_id: str) -> None: + """Reset and remove session state for a user.""" + with self._get_connection() as conn: + conn.execute("DELETE FROM whatsapp_sessions WHERE wa_id = ?", (wa_id,)) + logger.info(f"[WhatsApp] Session reset for {wa_id}") + + def is_message_processed(self, wamid: str) -> bool: + """Check whether a message ID has already been received and processed.""" + if not wamid: + return False + with self._get_connection() as conn: + cursor = conn.execute( + "SELECT 1 FROM processed_messages WHERE wamid = ?", + (wamid,) + ) + return cursor.fetchone() is not None + + def mark_message_processed(self, wamid: str) -> None: + """Record a message ID in the idempotency table.""" + if not wamid: + return + now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + with self._get_connection() as conn: + conn.execute( + "INSERT OR IGNORE INTO processed_messages (wamid, received_at) VALUES (?, ?)", + (wamid, now_str) + ) + + def cleanup_old_records(self, retention_days: int = 7) -> None: + """Purge idempotency records older than retention_days.""" + with self._get_connection() as conn: + conn.execute( + f"DELETE FROM processed_messages WHERE received_at < datetime('now', '-{retention_days} days')" + ) + + +# --------------------------------------------------------------------------- +# State Bridge: Supports both future Agent.get_state() and interim fallback +# --------------------------------------------------------------------------- + +def restore_agent_state(agent: Any, state: Optional[Dict[str, Any]]) -> None: + """Restore agent state. Prefers native agent.set_state() if present.""" + if not state: + return + + if hasattr(agent, "set_state") and callable(agent.set_state): + agent.set_state(state) + return + + # Fallback bridge before core SDialog implements Agent.set_state(): + memory_data = state.get("memory", []) + restored_msgs: List[BaseMessage] = [] + for item in memory_data: + msg_cls = _TYPE_TO_MSG_CLASS.get(item.get("type"), HumanMessage) + restored_msgs.append(msg_cls(content=item.get("content", ""))) + + if restored_msgs: + # Keep initial system prompt / persona prompt if not present in saved turns + if agent.memory and isinstance(agent.memory[0], SystemMessage): + agent.memory[:] = [agent.memory[0]] + restored_msgs + else: + agent.memory[:] = restored_msgs + + +def dump_agent_state(agent: Any, max_history_turns: int = 40) -> Dict[str, Any]: + """Serialize agent state. Prefers native agent.get_state() if present.""" + if hasattr(agent, "get_state") and callable(agent.get_state): + return agent.get_state() + + # Fallback bridge before core SDialog implements Agent.get_state(): + memory_items = [] + # Skip the primary system message (turn 0) in the dynamic history + history = agent.memory[1:] if (agent.memory and isinstance(agent.memory[0], SystemMessage)) else agent.memory + + # Apply sliding window budget + if len(history) > max_history_turns: + history = history[-max_history_turns:] + + for msg in history: + memory_items.append({ + "type": msg.__class__.__name__, + "content": msg.content + }) + + return { + "version": 1, + "memory": memory_items, + "finished": getattr(agent, "_finished", False) + } diff --git a/tests/test_whatsapp.py b/tests/test_whatsapp.py new file mode 100644 index 0000000..7a31ee8 --- /dev/null +++ b/tests/test_whatsapp.py @@ -0,0 +1,514 @@ +# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute +# SPDX-FileContributor: Sergio Burdisso +# SPDX-FileContributor: Amit Bhakt +# SPDX-License-Identifier: MIT +""" +Unit and integration tests for SDialog WhatsApp server. +""" + +import hmac +import hashlib +import json +import os +import tempfile +import time +import pytest +from unittest.mock import AsyncMock, patch, MagicMock + +from fastapi.testclient import TestClient +from langchain_core.messages import HumanMessage, AIMessage, SystemMessage + +from sdialog.whatsapp import ( + WhatsAppConfig, + WhatsAppServer, + SQLiteSessionStore, + MetaWhatsAppClient, +) +from sdialog.whatsapp.session_store import restore_agent_state, dump_agent_state + + +class MockWhatsAppAgent: + """Mock agent for WhatsApp testing.""" + + def __init__(self, name="TestBot"): + self.name = name + self.memory = [SystemMessage(content="You are a helpful test bot.")] + self.call_count = 0 + self.last_input = None + + def __call__(self, message: str, **kwargs) -> str: + self.call_count += 1 + self.last_input = message + self.memory.append(HumanMessage(content=message)) + reply = f"Bot reply to: {message}" + self.memory.append(AIMessage(content=reply)) + return reply + + def reset(self, seed=None): + self.memory = [SystemMessage(content="You are a helpful test bot.")] + self.call_count = 0 + + +@pytest.fixture +def temp_db(): + """Create a temporary SQLite database file for testing.""" + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + yield path + if os.path.exists(path): + os.remove(path) + + +@pytest.fixture +def test_config(temp_db): + """WhatsAppConfig configured for testing.""" + return WhatsAppConfig( + port=1234, + phone_number_id="1234567890", + access_token="test_access_token", + verify_token="test_verify_token", + app_secret="test_app_secret", + db_path=temp_db, + session_timeout=3600, + enforce_signature=True, + ) + + +# --------------------------------------------------------------------------- +# Test 1: Config +# --------------------------------------------------------------------------- + +def test_whatsapp_config_defaults(): + config = WhatsAppConfig(verify_token="custom_token") + assert config.port == 1234 + assert config.verify_token == "custom_token" + assert config.session_timeout == 1800 + assert config.max_history_turns == 40 + + +def test_whatsapp_config_validation(): + config = WhatsAppConfig(phone_number_id=None, access_token=None) + with pytest.raises(ValueError, match="phone_number_id is required"): + config.validate(dry_run=False) + + config = WhatsAppConfig(phone_number_id="123", access_token=None) + with pytest.raises(ValueError, match="access_token is required"): + config.validate(dry_run=False) + + # Valid config + config = WhatsAppConfig(phone_number_id="123", access_token="token") + config.validate(dry_run=False) # Should not raise + + +# --------------------------------------------------------------------------- +# Test 2: SQLiteSessionStore +# --------------------------------------------------------------------------- + +def test_session_store_save_load(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wa_id = "15551234567" + + # Initially empty + assert store.load_session(wa_id) is None + + # Save and reload + state = {"version": 1, "memory": [{"type": "HumanMessage", "content": "Hello"}]} + store.save_session(wa_id, state) + + loaded = store.load_session(wa_id) + assert loaded is not None + assert loaded["memory"][0]["content"] == "Hello" + + # Reset + store.reset_session(wa_id) + assert store.load_session(wa_id) is None + + +def test_session_store_timeout_expiration(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wa_id = "15551234567" + + state = {"version": 1, "memory": []} + store.save_session(wa_id, state) + + # With timeout_seconds=0, it should be expired immediately + time.sleep(0.01) + expired = store.load_session(wa_id, timeout_seconds=0) + assert expired is None + + +def test_session_store_idempotency(temp_db): + store = SQLiteSessionStore(db_path=temp_db) + wamid = "wamid.HBgLMTIzNDU2" + + assert not store.is_message_processed(wamid) + store.mark_message_processed(wamid) + assert store.is_message_processed(wamid) + + # Idempotent re-marking + store.mark_message_processed(wamid) + assert store.is_message_processed(wamid) + + +def test_state_bridge_restore_and_dump(): + agent = MockWhatsAppAgent() + agent.memory.append(HumanMessage(content="User 1")) + agent.memory.append(AIMessage(content="Bot 1")) + + # Dump state + state = dump_agent_state(agent, max_history_turns=10) + assert len(state["memory"]) == 2 + assert state["memory"][0]["content"] == "User 1" + assert state["memory"][1]["content"] == "Bot 1" + + # Restore on a fresh agent + new_agent = MockWhatsAppAgent() + restore_agent_state(new_agent, state) + assert len(new_agent.memory) == 3 # SystemMessage + 2 turns + assert isinstance(new_agent.memory[1], HumanMessage) + assert new_agent.memory[1].content == "User 1" + + +# --------------------------------------------------------------------------- +# Test 3: MetaWhatsAppClient +# --------------------------------------------------------------------------- + +def test_client_text_chunking(): + # Verify chunking does not break words prematurely + short_text = "Hello world" + assert MetaWhatsAppClient._chunk_text(short_text, 100) == [short_text] + + long_text = "First paragraph.\n\nSecond paragraph." + chunks = MetaWhatsAppClient._chunk_text(long_text, 20) + assert len(chunks) == 2 + assert chunks[0] == "First paragraph." + assert chunks[1] == "Second paragraph." + + +def test_client_send_text_mocked(test_config): + import asyncio + client = MetaWhatsAppClient(test_config) + + async def _test(): + with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_post.return_value = mock_response + + success = await client.send_text_message("15551234567", "Hello from test!") + assert success is True + assert mock_post.called + + called_url = mock_post.call_args[0][0] + assert "1234567890/messages" in called_url + + asyncio.run(_test()) + + +# --------------------------------------------------------------------------- +# Test 4: Webhook Verification (GET /webhook) +# --------------------------------------------------------------------------- + +def test_webhook_verification_success(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + response = client.get( + "/webhook", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "test_verify_token", + "hub.challenge": "test_challenge_code_123", + }, + ) + assert response.status_code == 200 + assert response.text == "test_challenge_code_123" + + +def test_webhook_verification_failure(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + response = client.get( + "/webhook", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "wrong_token", + "hub.challenge": "challenge", + }, + ) + assert response.status_code == 403 + + +# --------------------------------------------------------------------------- +# Test 5: Inbound Messages & HMAC (POST /webhook) +# --------------------------------------------------------------------------- + +def _create_meta_payload(sender="15551234567", text="Hello agent", wamid="wamid.123"): + """Helper to create a standard Meta WhatsApp webhook payload.""" + return { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "123456", + "changes": [ + { + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "metadata": { + "display_phone_number": "15550001", + "phone_number_id": "1234567890", + }, + "contacts": [ + { + "profile": {"name": "Alice"}, + "wa_id": sender, + } + ], + "messages": [ + { + "from": sender, + "id": wamid, + "timestamp": "1700000000", + "type": "text", + "text": {"body": text}, + } + ], + }, + } + ], + } + ], + } + + +def _compute_hmac(body: bytes, secret: str) -> str: + """Helper to compute Meta X-Hub-Signature-256 header value.""" + digest = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + return f"sha256={digest}" + + +def test_webhook_hmac_enforcement(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + payload = json.dumps(_create_meta_payload()).encode() + + # 1. Missing signature + resp = client.post("/webhook", content=payload, headers={"Content-Type": "application/json"}) + assert resp.status_code == 401 + + # 2. Invalid signature + resp = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": "sha256=invalid"}, + ) + assert resp.status_code == 401 + + # 3. Valid signature + valid_sig = _compute_hmac(payload, test_config.app_secret) + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + resp = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": valid_sig}, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "ok"} + + +def test_webhook_status_receipt_ignored(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + # Status receipt payload (e.g. message delivered) + status_payload = { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "123456", + "changes": [ + { + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "statuses": [{"id": "wamid.123", "status": "delivered"}], + }, + } + ], + } + ], + } + raw = json.dumps(status_payload).encode() + sig = _compute_hmac(raw, test_config.app_secret) + + resp = client.post( + "/webhook", + content=raw, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp.status_code == 200 + assert resp.json() == {"status": "ok"} + assert agent.call_count == 0 + + +def test_webhook_idempotency_duplicate_dropped(test_config): + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + payload = json.dumps(_create_meta_payload(wamid="wamid.repeat123")).encode() + sig = _compute_hmac(payload, test_config.app_secret) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + # First delivery: processed + resp1 = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp1.status_code == 200 + + # Second delivery (retry with same wamid): dropped + resp2 = client.post( + "/webhook", + content=payload, + headers={"Content-Type": "application/json", "X-Hub-Signature-256": sig}, + ) + assert resp2.status_code == 200 + assert agent.call_count == 1 # Agent called only once! + + +def test_webhook_multi_user_isolation(test_config): + """Verify that User A and User B have separate, isolated conversation histories.""" + agent = MockWhatsAppAgent() + app = WhatsAppServer.create_app(agent, test_config) + client = TestClient(app) + + with patch.object(MetaWhatsAppClient, "send_text_message", new_callable=AsyncMock): + # User A speaks + pA = json.dumps(_create_meta_payload(sender="1111", text="Hello from A", wamid="wamid.A1")).encode() + client.post( + "/webhook", + content=pA, + headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _compute_hmac(pA, test_config.app_secret), + }, + ) + + # User B speaks + pB = json.dumps(_create_meta_payload(sender="2222", text="Hello from B", wamid="wamid.B1")).encode() + client.post( + "/webhook", + content=pB, + headers={ + "Content-Type": "application/json", + "X-Hub-Signature-256": _compute_hmac(pB, test_config.app_secret), + }, + ) + + # Check SQLite store + store = SQLiteSessionStore(db_path=test_config.db_path) + stateA = store.load_session("1111") + stateB = store.load_session("2222") + + assert stateA is not None + assert stateB is not None + + # Verify User A's history contains User A's message, not User B's + contentsA = [m["content"] for m in stateA["memory"]] + contentsB = [m["content"] for m in stateB["memory"]] + + assert "Hello from A" in contentsA + assert "Hello from B" not in contentsA + + assert "Hello from B" in contentsB + assert "Hello from A" not in contentsB + + +def test_agent_serve_whatsapp_method_signature(): + """Verify serve_whatsapp is present on Agent and Server.""" + from sdialog.agents import Agent + from sdialog.server import Server + + assert hasattr(Agent, "serve_whatsapp") + assert callable(Agent.serve_whatsapp) + + assert hasattr(Server, "serve_whatsapp") + assert callable(Server.serve_whatsapp) + + +# --------------------------------------------------------------------------- +# Test 6: Auto-Discovery & Webhook Sync (adapted from WAVoice) +# --------------------------------------------------------------------------- + +def test_meta_discovery_inspect_and_discover(): + import asyncio + from sdialog.whatsapp.discovery import MetaDiscoveryClient + + client = MetaDiscoveryClient(access_token="test_token", app_secret="test_secret") + + async def _test(): + # Mock inspect_token + mock_token_data = { + "app_id": "123456", + "granular_scopes": [ + { + "scope": "whatsapp_business_management", + "target_ids": ["waba_101"], + } + ], + } + + with patch.object(MetaDiscoveryClient, "inspect_token", new_callable=AsyncMock) as mock_inspect: + mock_inspect.return_value = mock_token_data + + waba_id = await client.discover_waba_id() + assert waba_id == "waba_101" + + # Mock list_phone_numbers + with patch.object(MetaDiscoveryClient, "list_phone_numbers", new_callable=AsyncMock) as mock_list: + mock_list.return_value = [{"id": "phone_999", "display_phone_number": "+1555"}] + + phone_id = await client.discover_phone_number_id() + assert phone_id == "phone_999" + + asyncio.run(_test()) + + +def test_meta_discovery_configure_webhooks_mocked(): + import asyncio + from sdialog.whatsapp.discovery import MetaDiscoveryClient + + client = MetaDiscoveryClient(access_token="test_token", app_secret="test_secret") + + async def _test(): + with patch.object(MetaDiscoveryClient, "inspect_token", new_callable=AsyncMock) as mock_inspect: + mock_inspect.return_value = {"app_id": "app_555"} + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post: + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"success": True} + mock_post.return_value = mock_resp + + results = await client.configure_webhooks( + callback_url="https://test.ngrok.app/webhook", + verify_token="test_secret", + phone_number_id="phone_777", + waba_id="waba_888", + ) + + assert results["app_level"] is True + assert results["waba_level"] is True + assert results["phone_level"] is True + assert mock_post.call_count == 3 # All 3 tiers synced + + asyncio.run(_test())