mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-29 18:46:59 +00:00
fix(whatsapp): apply read receipts after intake policy
This commit is contained in:
Cyrus
kshitij
parent
35afa8ce06
commit
652d858f2e
6 changed files with 209 additions and 14 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue