feat(whatsapp): configurable inbound read receipts

gateway/config.py

@@ -1550,6 +1550,8 @@ def _merge_platform_map(source_platforms: Any) -> None:
                     bridged["cron_continuable_surface"] = platform_cfg["cron_continuable_surface"]
                 if "require_mention" in platform_cfg:
                     bridged["require_mention"] = platform_cfg["require_mention"]
+                if "send_read_receipts" in platform_cfg:
+                    bridged["send_read_receipts"] = platform_cfg["send_read_receipts"]
                 if plat == Platform.TELEGRAM and "allowed_chats" in platform_cfg:
                     bridged["allowed_chats"] = platform_cfg["allowed_chats"]
                 if plat == Platform.TELEGRAM and "group_allowed_chats" in platform_cfg:

plugins/platforms/whatsapp/adapter.py

@@ -379,6 +379,7 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
     - allow_from: List of sender IDs allowed in DMs (when dm_policy="allowlist")
     - group_policy: "open" | "allowlist" | "disabled" | "pairing" — which groups are processed (default: "pairing")
     - group_allow_from: List of group JIDs allowed (when group_policy="allowlist")
+    - send_read_receipts: Mark accepted inbound WhatsApp messages as read
 
     Behavior (gating, mention parsing, markdown conversion, chunking) is
     provided by ``WhatsAppBehaviorMixin`` so the Cloud API adapter can
@@ -410,6 +411,11 @@ def __init__(self, config: PlatformConfig):
         self._allow_from = self._coerce_allow_list(config.extra.get("allow_from") or config.extra.get("allowFrom"))
         self._group_policy = str(config.extra.get("group_policy") or os.getenv("WHATSAPP_GROUP_POLICY", "pairing")).strip().lower()
         self._group_allow_from = self._coerce_allow_list(config.extra.get("group_allow_from") or config.extra.get("groupAllowFrom"))
+        read_receipts = config.extra.get("send_read_receipts", False)
+        self._send_read_receipts = (
+            read_receipts if isinstance(read_receipts, bool)
+            else str(read_receipts or "").strip().lower() in {"1", "true", "yes", "on"}
+        )
         self._mention_patterns = self._compile_mention_patterns()
         self._message_queue: asyncio.Queue = asyncio.Queue()
         self._bridge_log_fh = None
@@ -592,17 +598,26 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
                                 # 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:
@@ -628,6 +643,9 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             bridge_env = with_hermes_node_path()
             if self._reply_prefix is not None:
                 bridge_env["WHATSAPP_REPLY_PREFIX"] = self._reply_prefix
+            bridge_env["WHATSAPP_SEND_READ_RECEIPTS"] = (
+                "true" if self._send_read_receipts else "false"
+            )
             # Pass the profile-aware cache directories so the bridge writes
             # media where the Python side reads it.  Without these the bridge
             # hardcodes ~/.hermes/{image,audio,document}_cache, which diverges
@@ -1253,6 +1271,10 @@ async def _poll_messages(self) -> None:
                         for msg_data in messages:
                             event = await self._build_message_event(msg_data)
                             if event:
+                                # Fire-and-forget: a slow bridge /read must not
+                                # delay message dispatch (matches BlueBubbles
+                                # asyncio.create_task pattern for mark_read).
+                                asyncio.create_task(self._send_read_receipt(msg_data))
                                 if event.message_type == MessageType.TEXT:
                                     self._enqueue_text_event(event)
                                 else:
@@ -1269,6 +1291,30 @@ async def _poll_messages(self) -> None:
 
             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

scripts/whatsapp-bridge/bridge.js

@@ -39,6 +39,7 @@ import {
   buildTextSendPayload,
   createBoundedMessageStore,
   extractBridgeEvent,
+  inboundReadReceiptKeys,
   inferMediaType,
   mediaPayloadForFile,
   pollCreationMessageFromPayload,
@@ -75,6 +76,12 @@ const FORWARD_OWNER_MESSAGES =
   typeof process.env.WHATSAPP_FORWARD_OWNER_MESSAGES === 'string' &&
   ['1', 'true', 'yes', 'on'].includes(process.env.WHATSAPP_FORWARD_OWNER_MESSAGES.toLowerCase());
 
+const SEND_READ_RECEIPTS =
+  typeof process !== 'undefined' &&
+  process.env &&
+  typeof process.env.WHATSAPP_SEND_READ_RECEIPTS === 'string' &&
+  ['1', 'true', 'yes', 'on'].includes(process.env.WHATSAPP_SEND_READ_RECEIPTS.toLowerCase());
+
 const PORT = parseInt(getArg('port', '3000'), 10);
 const SESSION_DIR = getArg('session', path.join(process.env.HOME || '~', '.hermes', 'whatsapp', 'session'));
 // Cache directories: the Python gateway passes the profile-aware paths via
@@ -1042,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;
@@ -1074,6 +1105,7 @@ app.get('/health', (req, res) => {
     queueLength: messageQueue.length,
     uptime: process.uptime(),
     scriptHash: SCRIPT_HASH,
+    sendReadReceipts: SEND_READ_RECEIPTS,
   });
 });
 

scripts/whatsapp-bridge/bridge.native.test.mjs

@@ -18,11 +18,33 @@ import {
   createBoundedMessageStore,
   appendMediaFailureNote,
   extractBridgeEvent,
+  inboundReadReceiptKeys,
   mediaPayloadForFile,
   pollCreationMessageFromPayload,
   pollUpdateForAggregation,
 } from './bridge_helpers.js';
 
+// -- inbound read receipts ------------------------------------------------
+{
+  const groupKey = {
+    id: 'incoming-group-1',
+    remoteJid: '120363001234567890@g.us',
+    participant: '15550001111@s.whatsapp.net',
+    fromMe: false,
+  };
+
+  assert.deepEqual(inboundReadReceiptKeys({ key: groupKey, enabled: false }), []);
+  assert.deepEqual(
+    inboundReadReceiptKeys({ key: { ...groupKey, fromMe: true }, enabled: true }),
+    [],
+  );
+  const receiptKeys = inboundReadReceiptKeys({ key: groupKey, enabled: true });
+  assert.equal(receiptKeys.length, 1);
+  assert.equal(receiptKeys[0], groupKey);
+  assert.equal(receiptKeys[0].participant, groupKey.participant);
+  console.log('  ✓ inbound read receipts preserve the original group message key');
+}
+
 // -- quoted outbound text -------------------------------------------------
 {
   const store = createBoundedMessageStore(2);
@@ -96,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');

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,
   };
 }
@@ -494,6 +500,12 @@ export function inferMediaType(ext) {
   return 'document';
 }
 
+export function inboundReadReceiptKeys({ key, enabled }) {
+  if (!enabled || !key || key.fromMe || !key.id || !key.remoteJid) return [];
+  // Preserve participant for group messages: Baileys needs the original key.
+  return [key];
+}
+
 export function mediaPayloadForFile({ buffer, filePath, mediaType, caption, fileName }) {
   const ext = filePath.toLowerCase().split('.').pop();
   const type = mediaType || inferMediaType(ext);

tests/gateway/test_whatsapp_connect.py

@@ -53,6 +53,7 @@ def _make_adapter():
     adapter._bridge_log = None
     adapter._bridge_process = None
     adapter._reply_prefix = None
+    adapter._send_read_receipts = False
     adapter._running = False
     adapter._message_handler = None
     adapter._fatal_error_code = None