| 1 | """ |
| 2 | WhatsApp handler — orchestrates poll, dispatch, and reply. |
| 3 | |
| 4 | Requires agent context. |
| 5 | """ |
| 6 | |
| 7 | import asyncio |
| 8 | import base64 |
| 9 | import os |
| 10 | import re |
| 11 | import uuid |
| 12 | |
| 13 | from agent import Agent, AgentContext, UserMessage |
| 14 | from helpers import plugins, files, runtime |
| 15 | from helpers import message_queue as mq |
| 16 | from helpers import integration_commands |
| 17 | from helpers.persist_chat import save_tmp_chat |
| 18 | from helpers.print_style import PrintStyle |
| 19 | from helpers.errors import format_error |
| 20 | from initialize import initialize_agent |
| 21 | |
| 22 | from plugins._whatsapp_integration.helpers import wa_client |
| 23 | from plugins._whatsapp_integration.helpers import bridge_manager |
| 24 | from plugins._whatsapp_integration.helpers.number_utils import ( |
| 25 | normalize_allowed_numbers, |
| 26 | normalize_number, |
| 27 | ) |
| 28 | |
| 29 | |
| 30 | PLUGIN_NAME = "_whatsapp_integration" |
| 31 | MEDIA_FOLDER = "usr/whatsapp/media" |
| 32 | |
| 33 | # Context data keys (no underscore prefix — must persist across restarts) |
| 34 | CTX_WA_CHAT_ID = "wa_chat_id" |
| 35 | CTX_WA_SENDER_NAME = "wa_sender_name" |
| 36 | CTX_WA_SENDER_NUMBER = "wa_sender_number" |
| 37 | CTX_WA_IS_GROUP = "wa_is_group" |
| 38 | CTX_WA_LAST_BODY = "wa_last_body" |
| 39 | CTX_WA_LAST_MSG_ID = "wa_last_msg_id" |
| 40 | # Transient — consumed per-reply, not persisted |
| 41 | CTX_WA_ATTACHMENTS = "_wa_response_attachments" |
| 42 | CTX_WA_REPLY_TO = "_wa_reply_to" |
| 43 | CTX_WA_TYPING_ACTIVE = "_wa_typing_active" |
| 44 | |
| 45 | # Poll task — lives here (not in extension module) because |
| 46 | # extension modules are re-executed on each job_loop tick, |
| 47 | # which would reset module-level state and orphan running tasks. |
| 48 | _poll_task: asyncio.Task | None = None # type: ignore[type-arg] |
| 49 | |
| 50 | |
| 51 | # ------------------------------------------------------------------ |
| 52 | # Poll loop |
| 53 | # ------------------------------------------------------------------ |
| 54 | |
| 55 | async def _refresh_typing(base_url: str) -> None: |
| 56 | """Re-send composing for all contexts with active typing flag.""" |
| 57 | for ctx in AgentContext._contexts.values(): |
| 58 | if not isinstance(ctx, AgentContext): |
| 59 | continue |
| 60 | if not ctx.data.get(CTX_WA_TYPING_ACTIVE): |
| 61 | continue |
| 62 | chat_id = ctx.data.get(CTX_WA_CHAT_ID, "") |
| 63 | if chat_id: |
| 64 | await wa_client.send_typing(base_url, chat_id) |
| 65 | |
| 66 | |
| 67 | async def poll_messages(config: dict) -> None: |
| 68 | if not config.get("enabled", False): |
| 69 | return |
| 70 | if PLUGIN_NAME not in plugins.get_enabled_plugins(None): |
| 71 | return |
| 72 | |
| 73 | port = int(config.get("bridge_port", 3100)) |
| 74 | base_url = bridge_manager.get_bridge_url(port) |
| 75 | |
| 76 | # Refresh typing indicator for active sessions (beats 25s WhatsApp timeout) |
| 77 | await _refresh_typing(base_url) |
| 78 | |
| 79 | try: |
| 80 | messages = await wa_client.get_messages(base_url) |
| 81 | except Exception as e: |
| 82 | PrintStyle.error(f"WhatsApp poll error: {format_error(e)}") |
| 83 | return |
| 84 | |
| 85 | if not messages: |
| 86 | return |
| 87 | |
| 88 | # Allowed-numbers filtering is authoritative in Python. |
| 89 | allowed_set = normalize_allowed_numbers(config.get("allowed_numbers")) |
| 90 | |
| 91 | for msg in messages: |
| 92 | try: |
| 93 | # Filter by allowed numbers if configured |
| 94 | if allowed_set: |
| 95 | sender_num = normalize_number(msg.get("senderNumber", "") or msg.get("senderId", "")) |
| 96 | if sender_num not in allowed_set: |
| 97 | PrintStyle.debug( |
| 98 | f"WhatsApp: ignored message from {sender_num} " |
| 99 | f"(senderId: {msg.get('senderId', '')}, allowed: {allowed_set})" |
| 100 | ) |
| 101 | continue |
| 102 | await _dispatch_message(config, msg) |
| 103 | except Exception as e: |
| 104 | PrintStyle.error(f"WhatsApp dispatch error: {format_error(e)}") |
| 105 | |
| 106 | |
| 107 | # ------------------------------------------------------------------ |
| 108 | # Dispatch a single inbound message |
| 109 | # ------------------------------------------------------------------ |
| 110 | |
| 111 | async def _dispatch_message(config: dict, msg: dict) -> None: |
| 112 | chat_id = msg.get("chatId", "") |
| 113 | is_group = msg.get("isGroup", False) |
| 114 | |
| 115 | # Group filtering: skip unless allow_group enabled AND bot was mentioned or replied to |
| 116 | if is_group: |
| 117 | if not config.get("allow_group", False): |
| 118 | PrintStyle.debug(f"WhatsApp: skipping group message (allow_group disabled)") |
| 119 | return |
| 120 | if not msg.get("mentionedMe", False) and not msg.get("repliedToMe", False): |
| 121 | PrintStyle.debug(f"WhatsApp: skipping group message (not mentioned or replied to)") |
| 122 | return |
| 123 | |
| 124 | existing = _find_chats_by_jid(chat_id) |
| 125 | |
| 126 | if existing: |
| 127 | # Continue most recent chat for this JID |
| 128 | if await _handle_control_message(config, msg, existing[0]): |
| 129 | return |
| 130 | port = int(config.get("bridge_port", 3100)) |
| 131 | base_url = bridge_manager.get_bridge_url(port) |
| 132 | await wa_client.send_typing(base_url, chat_id) |
| 133 | await _route_to_chat(msg, existing[0]) |
| 134 | else: |
| 135 | await _start_new_chat(config, msg) |
| 136 | |
| 137 | |
| 138 | # ------------------------------------------------------------------ |
| 139 | # Chat creation and routing |
| 140 | # ------------------------------------------------------------------ |
| 141 | |
| 142 | async def _start_new_chat(config: dict, msg: dict) -> None: |
| 143 | from helpers import projects |
| 144 | |
| 145 | sender_name = msg.get("senderName", "Unknown") |
| 146 | sender_number = msg.get("senderNumber", "") or normalize_number(msg.get("senderId", "")) |
| 147 | chat_id = msg.get("chatId", "") |
| 148 | is_group = msg.get("isGroup", False) |
| 149 | |
| 150 | agent_config = initialize_agent() |
| 151 | context = AgentContext(agent_config, name=f"WhatsApp: {sender_name[:50]}") |
| 152 | |
| 153 | context.data[CTX_WA_CHAT_ID] = chat_id |
| 154 | context.data[CTX_WA_SENDER_NAME] = sender_name |
| 155 | context.data[CTX_WA_SENDER_NUMBER] = sender_number |
| 156 | context.data[CTX_WA_IS_GROUP] = is_group |
| 157 | context.data[CTX_WA_LAST_BODY] = msg.get("body", "") |
| 158 | context.data[CTX_WA_LAST_MSG_ID] = msg.get("messageId", "") |
| 159 | context.data[CTX_WA_TYPING_ACTIVE] = True |
| 160 | |
| 161 | project = config.get("project", "") |
| 162 | if project: |
| 163 | projects.activate_project(context.id, project) |
| 164 | |
| 165 | save_tmp_chat(context) |
| 166 | |
| 167 | if await _handle_control_message(config, msg, context.id, context=context): |
| 168 | return |
| 169 | |
| 170 | port = int(config.get("bridge_port", 3100)) |
| 171 | base_url = bridge_manager.get_bridge_url(port) |
| 172 | await wa_client.send_typing(base_url, chat_id) |
| 173 | |
| 174 | user_msg = _build_user_message(context.agent0, msg) |
| 175 | system_ctx = context.agent0.read_prompt("fw.wa.system_context.md") |
| 176 | |
| 177 | msg_id = str(uuid.uuid4()) |
| 178 | media_urls = msg.get("mediaUrls", []) |
| 179 | attachments = await _save_incoming_media(media_urls) if media_urls else [] |
| 180 | mq.log_user_message( |
| 181 | context, user_msg, attachments, message_id=msg_id, source=" (whatsapp)", |
| 182 | ) |
| 183 | context.communicate(UserMessage( |
| 184 | message=user_msg, |
| 185 | system_message=[system_ctx], |
| 186 | attachments=attachments, |
| 187 | id=msg_id, |
| 188 | )) |
| 189 | |
| 190 | PrintStyle.success( |
| 191 | f"WhatsApp: new chat {context.id} for {sender_name} ({sender_number})" |
| 192 | ) |
| 193 | |
| 194 | |
| 195 | async def _route_to_chat( |
| 196 | msg: dict, context_id: str, |
| 197 | ) -> None: |
| 198 | context = AgentContext.get(context_id) |
| 199 | if not context: |
| 200 | return |
| 201 | |
| 202 | context.data[CTX_WA_LAST_BODY] = msg.get("body", "") |
| 203 | context.data[CTX_WA_LAST_MSG_ID] = msg.get("messageId", "") |
| 204 | context.data[CTX_WA_TYPING_ACTIVE] = True |
| 205 | |
| 206 | user_msg = _build_user_message(context.agent0, msg) |
| 207 | msg_id = str(uuid.uuid4()) |
| 208 | media_urls = msg.get("mediaUrls", []) |
| 209 | attachments = await _save_incoming_media(media_urls) if media_urls else [] |
| 210 | |
| 211 | if context.is_running(): |
| 212 | item = mq.add(context, user_msg, attachments) |
| 213 | save_tmp_chat(context) |
| 214 | port = int((plugins.get_plugin_config(PLUGIN_NAME) or {}).get("bridge_port", 3100)) |
| 215 | base_url = bridge_manager.get_bridge_url(port) |
| 216 | chat_id = context.data.get(CTX_WA_CHAT_ID, "") or msg.get("chatId", "") |
| 217 | reply_to = msg.get("messageId", "") if context.data.get(CTX_WA_IS_GROUP) else "" |
| 218 | await wa_client.send_message( |
| 219 | base_url, |
| 220 | chat_id, |
| 221 | f"Queued message #{item.get('seq', len(mq.get_queue(context)))}. Use /send to flush queued work, or /steer <message> to interrupt the active run.", |
| 222 | reply_to=reply_to, |
| 223 | ) |
| 224 | await wa_client.send_typing(base_url, chat_id, paused=True) |
| 225 | context.data[CTX_WA_TYPING_ACTIVE] = False |
| 226 | return |
| 227 | |
| 228 | mq.log_user_message( |
| 229 | context, user_msg, attachments, message_id=msg_id, source=" (whatsapp)", |
| 230 | ) |
| 231 | context.communicate(UserMessage( |
| 232 | message=user_msg, |
| 233 | attachments=attachments, |
| 234 | id=msg_id, |
| 235 | )) |
| 236 | |
| 237 | save_tmp_chat(context) |
| 238 | PrintStyle.info(f"WhatsApp: continuing chat {context_id}") |
| 239 | |
| 240 | |
| 241 | async def _handle_control_message( |
| 242 | config: dict, |
| 243 | msg: dict, |
| 244 | context_id: str, |
| 245 | *, |
| 246 | context: AgentContext | None = None, |
| 247 | ) -> bool: |
| 248 | text = msg.get("body", "") or "" |
| 249 | parsed = integration_commands.parse_command(text) |
| 250 | if not parsed: |
| 251 | return False |
| 252 | |
| 253 | context = context or AgentContext.get(context_id) |
| 254 | if not context: |
| 255 | return False |
| 256 | |
| 257 | response = integration_commands.try_handle_command(context, text) |
| 258 | if response is None: |
| 259 | return False |
| 260 | |
| 261 | port = int(config.get("bridge_port", 3100)) |
| 262 | base_url = bridge_manager.get_bridge_url(port) |
| 263 | chat_id = context.data.get(CTX_WA_CHAT_ID, "") or msg.get("chatId", "") |
| 264 | reply_to = msg.get("messageId", "") if context.data.get(CTX_WA_IS_GROUP) else "" |
| 265 | |
| 266 | await wa_client.send_message(base_url, chat_id, response, reply_to=reply_to) |
| 267 | await wa_client.send_typing(base_url, chat_id, paused=True) |
| 268 | context.data[CTX_WA_TYPING_ACTIVE] = False |
| 269 | PrintStyle.info(f"WhatsApp: handled control command in chat {context.id}") |
| 270 | return True |
| 271 | |
| 272 | |
| 273 | # ------------------------------------------------------------------ |
| 274 | # Chat discovery |
| 275 | # ------------------------------------------------------------------ |
| 276 | |
| 277 | def _find_chats_by_jid(chat_id: str) -> list[str]: |
| 278 | """Return context IDs for chats matching the given WhatsApp JID, newest first.""" |
| 279 | results = [] |
| 280 | for ctx_id, ctx in AgentContext._contexts.items(): |
| 281 | if not isinstance(ctx, AgentContext): |
| 282 | continue |
| 283 | if ctx.data.get(CTX_WA_CHAT_ID) != chat_id: |
| 284 | continue |
| 285 | results.append(ctx_id) |
| 286 | |
| 287 | results.sort(reverse=True) |
| 288 | return results |
| 289 | |
| 290 | |
| 291 | # ------------------------------------------------------------------ |
| 292 | # Markdown to WhatsApp formatting |
| 293 | # ------------------------------------------------------------------ |
| 294 | |
| 295 | def _md_to_whatsapp(text: str) -> str: |
| 296 | """Convert markdown formatting to WhatsApp formatting.""" |
| 297 | # Protect code blocks from conversion |
| 298 | code_blocks: list[str] = [] |
| 299 | def _save_code(m: re.Match) -> str: |
| 300 | code_blocks.append(m.group(0)) |
| 301 | return f"\x00CB{len(code_blocks) - 1}\x00" |
| 302 | text = re.sub(r"```[\s\S]*?```", _save_code, text) |
| 303 | |
| 304 | # Protect inline code |
| 305 | inline_codes: list[str] = [] |
| 306 | def _save_inline(m: re.Match) -> str: |
| 307 | inline_codes.append(m.group(0)) |
| 308 | return f"\x00IC{len(inline_codes) - 1}\x00" |
| 309 | text = re.sub(r"`[^`]+`", _save_inline, text) |
| 310 | |
| 311 | # Bold+italic ***text*** → *_text_* |
| 312 | text = re.sub(r"\*{3}(.+?)\*{3}", r"*_\1_*", text) |
| 313 | # Bold **text** or __text__ → *text* |
| 314 | text = re.sub(r"\*{2}(.+?)\*{2}", r"*\1*", text) |
| 315 | text = re.sub(r"__(.+?)__", r"*\1*", text) |
| 316 | # Italic _text_ stays _text_ (same in WhatsApp) |
| 317 | # Strikethrough ~~text~~ → ~text~ |
| 318 | text = re.sub(r"~~(.+?)~~", r"~\1~", text) |
| 319 | # Headings → bold |
| 320 | text = re.sub(r"^#{1,6}\s+(.+)$", r"*\1*", text, flags=re.MULTILINE) |
| 321 | |
| 322 | # Restore code blocks and inline code |
| 323 | for i, block in enumerate(code_blocks): |
| 324 | text = text.replace(f"\x00CB{i}\x00", block) |
| 325 | for i, code in enumerate(inline_codes): |
| 326 | text = text.replace(f"\x00IC{i}\x00", code) |
| 327 | |
| 328 | return text |
| 329 | |
| 330 | |
| 331 | # ------------------------------------------------------------------ |
| 332 | # Message builders |
| 333 | # ------------------------------------------------------------------ |
| 334 | |
| 335 | def _build_user_message(agent: Agent, msg: dict) -> str: |
| 336 | sender_name = msg.get("senderName", "Unknown") |
| 337 | sender_number = msg.get("senderNumber", "") or normalize_number(msg.get("senderId", "")) |
| 338 | is_group = msg.get("isGroup", False) |
| 339 | prompt = "fw.wa.user_message_group.md" if is_group else "fw.wa.user_message.md" |
| 340 | text = agent.read_prompt( |
| 341 | prompt, |
| 342 | sender_name=sender_name, |
| 343 | sender_number=sender_number, |
| 344 | group_name=msg.get("chatName", ""), |
| 345 | message_id=msg.get("messageId", ""), |
| 346 | body=msg.get("body", ""), |
| 347 | ) |
| 348 | return text |
| 349 | |
| 350 | |
| 351 | # ------------------------------------------------------------------ |
| 352 | # Reply sending (called from process_chain_end extension) |
| 353 | # ------------------------------------------------------------------ |
| 354 | |
| 355 | async def send_wa_reply( |
| 356 | context: AgentContext, |
| 357 | response_text: str, |
| 358 | attachments: list[str] | None = None, |
| 359 | reply_to: str = "", |
| 360 | keep_typing: bool = False, |
| 361 | ) -> str | None: |
| 362 | chat_id = context.data.get(CTX_WA_CHAT_ID) |
| 363 | if not chat_id: |
| 364 | return "No WhatsApp chat ID" |
| 365 | |
| 366 | config = plugins.get_plugin_config(PLUGIN_NAME) or {} |
| 367 | port = int(config.get("bridge_port", 3100)) |
| 368 | base_url = bridge_manager.get_bridge_url(port) |
| 369 | |
| 370 | # For group chats, auto-reply to last received message if no explicit reply_to |
| 371 | if not reply_to and context.data.get(CTX_WA_IS_GROUP): |
| 372 | reply_to = context.data.get(CTX_WA_LAST_MSG_ID, "") |
| 373 | |
| 374 | # Convert markdown to WhatsApp formatting |
| 375 | response_text = _md_to_whatsapp(response_text) |
| 376 | |
| 377 | # Prefix response in self-chat mode so user can distinguish agent messages |
| 378 | mode = config.get("mode", "self-chat") |
| 379 | if mode == "self-chat": |
| 380 | response_text = context.agent0.read_prompt( |
| 381 | "fw.wa.self_chat_prefix.md", response_text=response_text, |
| 382 | ) |
| 383 | |
| 384 | # Send text |
| 385 | try: |
| 386 | result = await wa_client.send_message(base_url, chat_id, response_text, reply_to=reply_to) |
| 387 | if result.get("error"): |
| 388 | return result["error"] |
| 389 | except Exception as e: |
| 390 | return str(e) |
| 391 | |
| 392 | # Send attachments via RFC (files may live in execution runtime) |
| 393 | if attachments: |
| 394 | host_paths = await _read_attachments_to_host(attachments) |
| 395 | for host_path in host_paths: |
| 396 | try: |
| 397 | result = await wa_client.send_media( |
| 398 | base_url, chat_id, host_path, |
| 399 | ) |
| 400 | if result.get("error"): |
| 401 | PrintStyle.warning(f"WhatsApp: attachment error: {result['error']}") |
| 402 | except Exception as e: |
| 403 | PrintStyle.warning(f"WhatsApp: attachment error: {e}") |
| 404 | |
| 405 | # Typing: restart if agent is still working, stop if final reply |
| 406 | if keep_typing: |
| 407 | await wa_client.send_typing(base_url, chat_id) |
| 408 | else: |
| 409 | await wa_client.send_typing(base_url, chat_id, paused=True) |
| 410 | context.data[CTX_WA_TYPING_ACTIVE] = False |
| 411 | |
| 412 | return None |
| 413 | |
| 414 | |
| 415 | # ------------------------------------------------------------------ |
| 416 | # Attachment reading (via RFC into execution runtime) |
| 417 | # ------------------------------------------------------------------ |
| 418 | |
| 419 | async def _read_attachments_to_host( |
| 420 | paths: list[str], |
| 421 | ) -> list[str]: |
| 422 | """Read files from execution runtime and write to host media cache.""" |
| 423 | from plugins._whatsapp_integration.helpers.attachment_reader import read_attachment |
| 424 | |
| 425 | host_paths: list[str] = [] |
| 426 | for path in paths: |
| 427 | data = await runtime.call_development_function(read_attachment, path) |
| 428 | if data["error"]: |
| 429 | PrintStyle.warning(f"WhatsApp attachment: {data['error']}") |
| 430 | continue |
| 431 | # Write decoded bytes to host-side media cache |
| 432 | host_path = os.path.join( |
| 433 | files.get_abs_path(MEDIA_FOLDER), data["name"], |
| 434 | ) |
| 435 | os.makedirs(os.path.dirname(host_path), exist_ok=True) |
| 436 | with open(host_path, "wb") as f: |
| 437 | f.write(base64.b64decode(data["content_b64"])) |
| 438 | host_paths.append(host_path) |
| 439 | return host_paths |
| 440 | |
| 441 | |
| 442 | async def _save_incoming_media( |
| 443 | media_urls: list[str], |
| 444 | ) -> list[str]: |
| 445 | """Save incoming media files into execution runtime via RFC.""" |
| 446 | from plugins._whatsapp_integration.helpers.attachment_writer import write_attachment |
| 447 | |
| 448 | runtime_paths: list[str] = [] |
| 449 | for host_path in media_urls: |
| 450 | if not os.path.isfile(host_path): |
| 451 | continue |
| 452 | name = os.path.basename(host_path) |
| 453 | with open(host_path, "rb") as f: |
| 454 | content_b64 = base64.b64encode(f.read()).decode() |
| 455 | rel_path = os.path.join(MEDIA_FOLDER, name) |
| 456 | result = await runtime.call_development_function( |
| 457 | write_attachment, rel_path, content_b64, |
| 458 | ) |
| 459 | if result.get("error"): |
| 460 | PrintStyle.warning(f"WhatsApp media save: {result['error']}") |
| 461 | runtime_paths.append(host_path) # fallback to host path |
| 462 | else: |
| 463 | runtime_paths.append(result["path"]) |
| 464 | return runtime_paths |