diff --git a/plugins/platforms/whatsapp/adapter.py b/plugins/platforms/whatsapp/adapter.py index b77c9eb52d26..62689281b50f 100644 --- a/plugins/platforms/whatsapp/adapter.py +++ b/plugins/platforms/whatsapp/adapter.py @@ -598,17 +598,26 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter): # treated as stale by definition. running_hash = data.get("scriptHash", "") disk_hash = _file_content_hash(bridge_path) - if running_hash and disk_hash and running_hash == disk_hash: + running_read_receipts = bool(data.get("sendReadReceipts", False)) + config_matches = running_read_receipts == self._send_read_receipts + if ( + running_hash + and disk_hash + and running_hash == disk_hash + and config_matches + ): print(f"[{self.name}] Using existing bridge (status: {bridge_status})") self._mark_connected() self._bridge_process = None # Not managed by us self._http_session = aiohttp.ClientSession() self._poll_task = asyncio.create_task(self._poll_messages()) return True - print( - f"[{self.name}] Running bridge is stale " - f"(running={running_hash or 'unversioned'}, disk={disk_hash}), restarting" + stale_reason = ( + f"running={running_hash or 'unversioned'}, disk={disk_hash}" + if running_hash != disk_hash + else "send_read_receipts config changed" ) + print(f"[{self.name}] Running bridge is stale ({stale_reason}), restarting") else: print(f"[{self.name}] Bridge found but not connected (status: {bridge_status}), restarting") except Exception: @@ -1262,6 +1271,7 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter): for msg_data in messages: event = await self._build_message_event(msg_data) if event: + await self._send_read_receipt(msg_data) if event.message_type == MessageType.TEXT: self._enqueue_text_event(event) else: @@ -1278,6 +1288,30 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter): await asyncio.sleep(1) # Poll interval + async def _send_read_receipt(self, data: Dict[str, Any]) -> None: + """Mark a policy-accepted inbound message as read via the bridge.""" + if not self._send_read_receipts or not self._http_session: + return + key = data.get("readReceiptKey") + if not isinstance(key, dict): + return + try: + import aiohttp + + async with self._http_session.post( + f"http://127.0.0.1:{self._bridge_port}/read", + json={"key": key}, + timeout=aiohttp.ClientTimeout(total=5), + ) as resp: + if resp.status != 200: + logger.warning( + "[%s] WhatsApp read receipt failed with HTTP %s", + self.name, + resp.status, + ) + except Exception as exc: + logger.warning("[%s] WhatsApp read receipt failed: %s", self.name, exc) + # ── Text debounce batching ────────────────────────────────────── _SPLIT_THRESHOLD = 6000 # WhatsApp supports ~65K chars; generous threshold diff --git a/scripts/whatsapp-bridge/bridge.js b/scripts/whatsapp-bridge/bridge.js index 500879a20bfb..fbacb6b43f8a 100644 --- a/scripts/whatsapp-bridge/bridge.js +++ b/scripts/whatsapp-bridge/bridge.js @@ -654,15 +654,6 @@ async function startSocket() { } } - const receiptKeys = inboundReadReceiptKeys({ key: msg.key, enabled: SEND_READ_RECEIPTS }); - if (receiptKeys.length > 0) { - try { - await sock.readMessages(receiptKeys); - } catch (err) { - console.warn('[bridge] failed to send read receipt:', err.message); - } - } - const messageContent = getMessageContent(msg); if (messageContent.pollUpdateMessage) { const pollUpdateMessage = messageContent.pollUpdateMessage; @@ -1058,6 +1049,30 @@ app.post('/typing', async (req, res) => { } }); +// Mark an inbound message as read only after the Python adapter has accepted +// it through the authoritative DM/group/mention intake policy. +app.post('/read', async (req, res) => { + if (!sock || connectionState !== 'connected') { + return res.status(503).json({ error: 'Not connected' }); + } + + const receiptKeys = inboundReadReceiptKeys({ + key: req.body?.key, + enabled: SEND_READ_RECEIPTS, + }); + if (receiptKeys.length === 0) { + return res.json({ success: true, marked: false }); + } + + try { + await sock.readMessages(receiptKeys); + return res.json({ success: true, marked: true }); + } catch (err) { + console.warn('[bridge] failed to send read receipt:', err.message); + return res.status(500).json({ error: 'Failed to send read receipt' }); + } +}); + // Chat info app.get('/chat/:id', async (req, res) => { const chatId = req.params.id; diff --git a/scripts/whatsapp-bridge/bridge.native.test.mjs b/scripts/whatsapp-bridge/bridge.native.test.mjs index b31b95c46c30..b4e123a265ef 100644 --- a/scripts/whatsapp-bridge/bridge.native.test.mjs +++ b/scripts/whatsapp-bridge/bridge.native.test.mjs @@ -118,6 +118,12 @@ import { assert.equal(event.quotedParticipant, '15559998888@s.whatsapp.net'); assert.equal(event.quotedRemoteJid, '15551234567@s.whatsapp.net'); assert.equal(event.quotedText, 'approve deploy?'); + assert.deepEqual(event.readReceiptKey, { + id: 'incoming-1', + remoteJid: '15551234567@s.whatsapp.net', + participant: '15550001111@s.whatsapp.net', + fromMe: false, + }); assert.equal(event.hasQuotedMessage, true); assert.equal(event.body, 'approved'); console.log(' ✓ inbound quoted metadata includes quoted text'); diff --git a/scripts/whatsapp-bridge/bridge_helpers.js b/scripts/whatsapp-bridge/bridge_helpers.js index fd5444d80462..c0618cd2bd6e 100644 --- a/scripts/whatsapp-bridge/bridge_helpers.js +++ b/scripts/whatsapp-bridge/bridge_helpers.js @@ -483,6 +483,12 @@ export async function extractBridgeEvent({ quotedText, hasQuotedMessage, botIds, + readReceiptKey: { + remoteJid: msg.key.remoteJid || chatId, + id: msg.key.id, + participant: msg.key.participant || senderId, + fromMe: Boolean(msg.key.fromMe), + }, timestamp: msg.messageTimestamp, }; } diff --git a/tests/gateway/test_whatsapp_reply_prefix.py b/tests/gateway/test_whatsapp_reply_prefix.py index 42383e42d14e..ba6084096ba6 100644 --- a/tests/gateway/test_whatsapp_reply_prefix.py +++ b/tests/gateway/test_whatsapp_reply_prefix.py @@ -9,10 +9,25 @@ Covers: - Config version covers all ENV_VARS_BY_VERSION keys (regression guard) """ -from unittest.mock import patch +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch +import pytest from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType + + +class _AsyncResponseContext: + def __init__(self, response): + self.response = response + + async def __aenter__(self): + return self.response + + async def __aexit__(self, exc_type, exc, tb): + return False # --------------------------------------------------------------------------- @@ -134,6 +149,91 @@ class TestAdapterInit: )._send_read_receipts is False +class TestReadReceiptPolicyOrdering: + @pytest.mark.asyncio + async def test_accepted_receipt_key_is_sent_to_bridge(self): + from plugins.platforms.whatsapp.adapter import WhatsAppAdapter + + adapter = WhatsAppAdapter( + PlatformConfig(enabled=True, extra={"send_read_receipts": True}) + ) + response = SimpleNamespace(status=200) + session = MagicMock() + session.post.return_value = _AsyncResponseContext(response) + adapter._http_session = session + key = { + "id": "incoming-1", + "remoteJid": "120363001234567890@g.us", + "participant": "15550001111@s.whatsapp.net", + "fromMe": False, + } + + await adapter._send_read_receipt({"readReceiptKey": key}) + + assert session.post.call_args.kwargs["json"] == {"key": key} + assert session.post.call_args.args[0].endswith("/read") + + @pytest.mark.asyncio + async def test_rejected_message_is_not_marked_read(self, monkeypatch): + from plugins.platforms.whatsapp.adapter import WhatsAppAdapter + + adapter = WhatsAppAdapter( + PlatformConfig(enabled=True, extra={"send_read_receipts": True}) + ) + response = SimpleNamespace( + status=200, + json=AsyncMock(return_value=[{"messageId": "ignored"}]), + ) + session = MagicMock() + session.get.return_value = _AsyncResponseContext(response) + adapter._http_session = session + adapter._running = True + adapter._check_managed_bridge_exit = AsyncMock(return_value=None) + adapter._send_read_receipt = AsyncMock() + + async def _reject(data): + adapter._running = False + return None + + adapter._build_message_event = _reject + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + await adapter._poll_messages() + + adapter._send_read_receipt.assert_not_awaited() + + @pytest.mark.asyncio + async def test_policy_accepted_message_is_marked_read_before_dispatch(self, monkeypatch): + from plugins.platforms.whatsapp.adapter import WhatsAppAdapter + + adapter = WhatsAppAdapter( + PlatformConfig(enabled=True, extra={"send_read_receipts": True}) + ) + raw = {"messageId": "accepted"} + response = SimpleNamespace(status=200, json=AsyncMock(return_value=[raw])) + session = MagicMock() + session.get.return_value = _AsyncResponseContext(response) + adapter._http_session = session + adapter._running = True + adapter._check_managed_bridge_exit = AsyncMock(return_value=None) + adapter._send_read_receipt = AsyncMock() + adapter.handle_message = AsyncMock() + event = MagicMock(spec=MessageEvent) + event.message_type = MessageType.PHOTO + + async def _accept(data): + adapter._running = False + return event + + adapter._build_message_event = _accept + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + await adapter._poll_messages() + + adapter._send_read_receipt.assert_awaited_once_with(raw) + adapter.handle_message.assert_awaited_once_with(event) + + # --------------------------------------------------------------------------- # Config version regression guard # --------------------------------------------------------------------------- diff --git a/tests/gateway/test_whatsapp_stale_bridge.py b/tests/gateway/test_whatsapp_stale_bridge.py index ccc920c5630a..65cda1ce2f24 100644 --- a/tests/gateway/test_whatsapp_stale_bridge.py +++ b/tests/gateway/test_whatsapp_stale_bridge.py @@ -197,6 +197,40 @@ class TestStaleBridgeHandshake: mock_popen.assert_called_once() # stale bridge replaced, not reused mock_kill_port.assert_called_once_with(adapter._bridge_port) + @pytest.mark.asyncio + async def test_restarts_bridge_when_read_receipt_config_changed(self, tmp_path): + from plugins.platforms.whatsapp.adapter import _file_content_hash + + bridge_dir = _setup_bridge_dir(tmp_path) + _fresh_node_modules(bridge_dir) + adapter = _make_adapter( + bridge_script=str(bridge_dir / "bridge.js"), + session_path=tmp_path / "session", + ) + adapter._send_read_receipts = True + disk_hash = _file_content_hash(bridge_dir / "bridge.js") + mock_client = _mock_health( + { + "status": "connected", + "scriptHash": disk_hash, + "sendReadReceipts": False, + } + ) + mock_proc = MagicMock() + mock_proc.poll.return_value = 1 + mock_proc.returncode = 1 + + with patch("plugins.platforms.whatsapp.adapter.check_whatsapp_requirements", return_value=True), \ + patch("aiohttp.ClientSession", mock_client), \ + patch("plugins.platforms.whatsapp.adapter.asyncio.sleep", new_callable=AsyncMock), \ + patch("plugins.platforms.whatsapp.adapter._kill_stale_bridge_by_pidfile"), \ + patch("plugins.platforms.whatsapp.adapter._kill_port_process"), \ + patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \ + patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True): + await adapter.connect() + + mock_popen.assert_called_once() + @pytest.mark.asyncio async def test_restarts_unversioned_bridge(self, tmp_path): """Bridges predating the handshake report no scriptHash → stale."""