diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index c2213daeef1e7..f2926e587c79a 100644 --- a/gateway/platforms/bluebubbles.py +++ b/gateway/platforms/bluebubbles.py @@ -9,10 +9,12 @@ """ import asyncio +import hashlib import json import logging import os import re +import time import uuid from collections import OrderedDict from datetime import datetime @@ -65,6 +67,22 @@ # Webhook event types that carry user messages _MESSAGE_EVENTS = {"new-message", "message", "updated-message"} +# Register only new user messages by default. BlueBubbles also emits +# updated-message for receipt/status metadata, and some deployments dispatch +# the same inbound iMessage through both paths. Advanced users can explicitly +# opt into updated-message via webhook_events / BLUEBUBBLES_WEBHOOK_EVENTS; the +# handler below still ACKs status-only updates without starting an agent turn. +_DEFAULT_WEBHOOK_EVENTS = ["new-message"] +_VALID_WEBHOOK_EVENTS = { + "new-message", + "updated-message", + "message-send-error", + "group-name-change", + "participant-added", + "participant-removed", +} +_DEDUP_TTL_SECONDS = 60.0 +_DEDUP_MAX_ENTRIES = 1000 # Log redaction patterns _PHONE_RE = re.compile(r"\+?\d{7,15}") @@ -150,11 +168,155 @@ def __init__(self, config: PlatformConfig): self._private_api_enabled: Optional[bool] = None self._helper_connected: bool = False self._guid_cache: OrderedDict[str, str] = OrderedDict() + self.webhook_events = self._configured_webhook_events(extra) + self._seen_inbound_messages: OrderedDict[str, float] = OrderedDict() + self._owns_webhook_listener = False # ------------------------------------------------------------------ # API helpers # ------------------------------------------------------------------ + @staticmethod + def _configured_webhook_events(extra: Dict[str, Any]) -> List[str]: + raw = extra.get("webhook_events") or os.getenv("BLUEBUBBLES_WEBHOOK_EVENTS") + if raw is None: + return list(_DEFAULT_WEBHOOK_EVENTS) + if isinstance(raw, str): + candidates = [item.strip() for item in raw.split(",")] + elif isinstance(raw, (list, tuple, set)): + candidates = [str(item).strip() for item in raw] + else: + logger.warning( + "[bluebubbles] invalid webhook_events config %r; using default %s", + raw, + _DEFAULT_WEBHOOK_EVENTS, + ) + return list(_DEFAULT_WEBHOOK_EVENTS) + + events: List[str] = [] + for event in candidates: + if not event: + continue + if event not in _VALID_WEBHOOK_EVENTS: + logger.warning("[bluebubbles] ignoring unsupported webhook event: %s", event) + continue + if event not in events: + events.append(event) + return events or list(_DEFAULT_WEBHOOK_EVENTS) + + @staticmethod + def _message_dedup_key(payload: Dict[str, Any], record: Dict[str, Any], text: str) -> str: + message_guid = BlueBubblesAdapter._value( + record.get("guid"), + record.get("messageGuid"), + record.get("id"), + payload.get("messageGuid"), + ) + chat_guid = BlueBubblesAdapter._value( + record.get("chatGuid"), + payload.get("chatGuid"), + record.get("chat_guid"), + payload.get("chat_guid"), + ) + if not chat_guid: + chats = record.get("chats") or [] + if chats and isinstance(chats[0], dict): + chat_guid = chats[0].get("guid") or chats[0].get("chatGuid") + date_created = BlueBubblesAdapter._value( + str(record.get("dateCreated")) if record.get("dateCreated") is not None else None, + str(record.get("date_created")) if record.get("date_created") is not None else None, + ) + text_hash = hashlib.sha256(text.encode("utf-8", errors="ignore")).hexdigest() + if message_guid: + return f"guid:{message_guid}:{text_hash}" + return f"fallback:{chat_guid or ''}:{date_created or ''}:{text_hash}" + + @staticmethod + def _message_content_dedup_key(payload: Dict[str, Any], record: Dict[str, Any], text: str) -> str: + """Best-effort duplicate key for BlueBubbles' duplicate DM webhooks. + + Some BlueBubbles deployments deliver the same DM twice with different + chat IDs (for example ``any;-;+155...`` and ``+155...``) and sometimes + different message GUIDs. The normal GUID-based key cannot catch that, + so keep a short-lived canonical sender/chat/text key as a second guard. + """ + chat_guid = BlueBubblesAdapter._value( + record.get("chatGuid"), + payload.get("chatGuid"), + record.get("chat_guid"), + payload.get("chat_guid"), + ) + chats = record.get("chats") or [] + chat_identifier = BlueBubblesAdapter._value( + record.get("chatIdentifier"), + record.get("identifier"), + payload.get("chatIdentifier"), + payload.get("identifier"), + ) + if not chat_guid and chats and isinstance(chats[0], dict): + chat_guid = chats[0].get("guid") or chats[0].get("chatGuid") + chat_identifier = chat_identifier or chats[0].get("chatIdentifier") + sender = ( + BlueBubblesAdapter._value( + record.get("handle", {}).get("address") + if isinstance(record.get("handle"), dict) + else None, + record.get("sender"), + record.get("from"), + record.get("address"), + ) + or chat_identifier + or chat_guid + ) + is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or "")) + canonical_chat = chat_guid if is_group else BlueBubblesAdapter._canonical_dm_chat_identifier( + chat_guid, + chat_identifier, + sender, + ) + text_hash = hashlib.sha256(text.encode("utf-8", errors="ignore")).hexdigest() + return f"recent:{'group' if is_group else 'dm'}:{canonical_chat}:{sender or ''}:{text_hash}" + + @staticmethod + def _canonical_dm_chat_identifier( + chat_guid: Optional[str], + chat_identifier: Optional[str], + sender: Optional[str], + ) -> str: + """Normalize BlueBubbles DM variants to one session/dedup identity.""" + for candidate in (chat_identifier, sender): + if candidate and ";" not in candidate: + return candidate + if chat_guid and ";-;" in chat_guid: + suffix = chat_guid.rsplit(";-;", 1)[-1].strip() + if suffix: + return suffix + return chat_identifier or sender or chat_guid or "" + + def _is_duplicate_inbound_message(self, *keys: str) -> bool: + now = time.monotonic() + while self._seen_inbound_messages: + _, seen_at = next(iter(self._seen_inbound_messages.items())) + if now - seen_at <= _DEDUP_TTL_SECONDS and len(self._seen_inbound_messages) <= _DEDUP_MAX_ENTRIES: + break + self._seen_inbound_messages.popitem(last=False) + + duplicate = False + for key in keys: + if not key: + continue + if key in self._seen_inbound_messages: + seen_at = self._seen_inbound_messages[key] + self._seen_inbound_messages.move_to_end(key) + self._seen_inbound_messages[key] = now + duplicate = duplicate or (now - seen_at <= _DEDUP_TTL_SECONDS) + else: + self._seen_inbound_messages[key] = now + + while len(self._seen_inbound_messages) > _DEDUP_MAX_ENTRIES: + self._seen_inbound_messages.popitem(last=False) + return duplicate + def _api_url(self, path: str) -> str: sep = "&" if "?" in path else "?" return f"{self.server_url}{path}{sep}password={quote(self.password, safe='')}" @@ -263,6 +425,20 @@ async def connect(self) -> bool: self.client = None return False + listener_started = await self._start_webhook_listener(web) + self._owns_webhook_listener = listener_started + self._mark_connected() + + if listener_started: + # Register webhook with BlueBubbles server only when this adapter owns + # the local listener. One-shot outbound senders may run while the + # gateway already owns the same webhook host/port; those senders can + # still deliver via REST and should not fail on EADDRINUSE. + await self._register_webhook() + + return True + + async def _start_webhook_listener(self, web) -> bool: app = web.Application() app.router.add_get("/health", lambda _: web.Response(text="ok")) app.router.add_post(self.webhook_path, self._handle_webhook) @@ -271,25 +447,37 @@ async def connect(self) -> bool: # aiohttp access logs write that request target to agent.log. self._runner = web.AppRunner(app, access_log=None) await self._runner.setup() - site = web.TCPSite(self._runner, self.webhook_host, self.webhook_port) - await site.start() - self._mark_connected() + try: + site = web.TCPSite(self._runner, self.webhook_host, self.webhook_port) + await site.start() + except OSError as exc: + await self._runner.cleanup() + self._runner = None + if getattr(exc, "errno", None) == 98 or "address already in use" in str(exc).lower(): + logger.warning( + "[bluebubbles] webhook listener already in use on %s:%s; " + "continuing in outbound-only mode", + self.webhook_host, + self.webhook_port, + ) + return False + raise logger.info( "[bluebubbles] webhook listening on http://%s:%s%s", self.webhook_host, self.webhook_port, self.webhook_path, ) - - # Register webhook with BlueBubbles server - # This is required for the server to know where to send events - await self._register_webhook() - return True async def disconnect(self) -> None: - # Unregister webhook before cleaning up - await self._unregister_webhook() + # Unregister webhook only when this adapter owns the local listener. + # Outbound-only one-shot senders may reuse the REST client while the + # long-running gateway owns the listener; they must not remove that + # gateway's webhook registration during cleanup. + if self._owns_webhook_listener: + await self._unregister_webhook() + self._owns_webhook_listener = False if self.client: await self.client.aclose() @@ -353,18 +541,36 @@ async def _register_webhook(self) -> bool: webhook_url = self._webhook_register_url - # Crash resilience — reuse an existing registration if present + # Crash resilience — reuse an existing registration if present, but + # replace registrations whose event list differs from the adapter's + # desired receipt/edit-capable set. existing = await self._find_registered_webhooks(webhook_url) if existing: - logger.info( - "[bluebubbles] webhook already registered: %s", - self._webhook_register_url_for_log, - ) - return True + desired_events = list(self.webhook_events) + if any(wh.get("events") == desired_events for wh in existing): + logger.info( + "[bluebubbles] webhook already registered: %s", + self._webhook_register_url_for_log, + ) + return True + for wh in existing: + wh_id = wh.get("id") + if wh_id: + try: + res = await self.client.delete( + self._api_url(f"/api/v1/webhook/{wh_id}") + ) + res.raise_for_status() + except Exception as exc: + logger.debug( + "[bluebubbles] failed to remove stale webhook %s: %s", + wh_id, + exc, + ) payload = { "url": webhook_url, - "events": ["new-message", "updated-message"], + "events": list(self.webhook_events), } try: @@ -506,16 +712,13 @@ async def send( text = self.format_message(content) if not text: return SendResult(success=False, error="BlueBubbles send requires text") - # Split on paragraph breaks first (double newlines) so each thought - # becomes its own iMessage bubble, then truncate any that are still - # too long. - paragraphs = [p.strip() for p in re.split(r'\n\s*\n', text) if p.strip()] - chunks: List[str] = [] - for para in (paragraphs or [text]): - if len(para) <= self.MAX_MESSAGE_LENGTH: - chunks.append(para) - else: - chunks.extend(self.truncate_message(para, max_length=self.MAX_MESSAGE_LENGTH)) + # Send normal replies as a single iMessage bubble. Splitting on blank + # lines made one assistant response look like multiple/duplicate replies + # in BlueBubbles; only chunk when the platform text limit requires it. + if len(text) <= self.MAX_MESSAGE_LENGTH: + chunks = [text] + else: + chunks = self.truncate_message(text, max_length=self.MAX_MESSAGE_LENGTH) last = SendResult(success=True) for chunk in chunks: guid = await self._resolve_chat_guid(chat_id) @@ -715,7 +918,16 @@ async def stop_typing(self, chat_id: str) -> None: # ------------------------------------------------------------------ async def mark_read(self, chat_id: str) -> bool: + if not self.send_read_receipts: + logger.debug("[bluebubbles] mark_read skipped: send_read_receipts disabled") + return False if not self._private_api_enabled or not self._helper_connected or not self.client: + logger.debug( + "[bluebubbles] mark_read skipped: private_api=%s helper=%s client=%s", + self._private_api_enabled, + self._helper_connected, + bool(self.client), + ) return False try: guid = await self._resolve_chat_guid(chat_id) @@ -724,9 +936,15 @@ async def mark_read(self, chat_id: str) -> bool: await self.client.post( self._api_url(f"/api/v1/chat/{encoded}/read"), timeout=5 ) + logger.info("[bluebubbles] marked chat read: %s", _redact(guid)) return True - except Exception: - pass + logger.debug("[bluebubbles] mark_read skipped: no guid for %s", _redact(chat_id)) + except Exception as exc: + logger.warning( + "[bluebubbles] failed to mark chat read for %s: %s", + _redact(chat_id), + exc, + ) return False # ------------------------------------------------------------------ @@ -896,6 +1114,8 @@ async def _handle_webhook(self, request): # Only process message events; silently acknowledge everything else if event_type and event_type not in _MESSAGE_EVENTS: return web.Response(text="ok") + if event_type and event_type not in self.webhook_events: + return web.Response(text="ok") record = self._extract_payload_record(payload) or {} is_from_me = bool( @@ -921,6 +1141,18 @@ async def _handle_webhook(self, request): or "" ) + if event_type == "updated-message": + has_edit = bool(record.get("dateEdited") or record.get("date_edited")) + has_retraction = bool( + record.get("dateRetracted") or record.get("date_retracted") + ) + # BlueBubbles also emits updated-message for delivery/read receipts + # and other status-only metadata changes. Those should keep the + # webhook subscription alive for receipt visibility, but must not be + # routed to the agent as duplicate user messages. + if not has_edit and not has_retraction: + return web.Response(text="ok") + # --- Inbound attachment handling --- attachments = record.get("attachments") or [] media_urls: List[str] = [] @@ -957,6 +1189,12 @@ async def _handle_webhook(self, request): text = "(attachment)" # --- End attachment handling --- + dedup_key = self._message_dedup_key(payload, record, text) + content_dedup_key = self._message_content_dedup_key(payload, record, text) + if self._is_duplicate_inbound_message(dedup_key, content_dedup_key): + logger.info("[bluebubbles] duplicate inbound webhook ignored") + return web.Response(text="ok") + chat_guid = self._value( record.get("chatGuid"), payload.get("chatGuid"), @@ -993,8 +1231,15 @@ async def _handle_webhook(self, request): if not sender or not (chat_guid or chat_identifier) or not text: return web.json_response({"error": "missing message fields"}, status=400) - session_chat_id = chat_guid or chat_identifier is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or "")) + if is_group: + session_chat_id = chat_guid or chat_identifier + else: + session_chat_id = self._canonical_dm_chat_identifier( + chat_guid, + chat_identifier, + sender, + ) if is_group and self.require_mention: if not self._message_matches_mention_patterns(text): logger.debug( diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 7d4a71378c0bf..275081f4b37b7 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/tests/gateway/test_bluebubbles.py @@ -10,6 +10,7 @@ def _make_adapter(monkeypatch, **extra): monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234") monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret") + monkeypatch.setenv("BLUEBUBBLES_WEBHOOK_HOST", "127.0.0.1") from gateway.platforms.bluebubbles import BlueBubblesAdapter cfg = PlatformConfig( @@ -85,7 +86,7 @@ def test_truncate_message_omits_pagination_suffixes(self, monkeypatch): assert all("(" not in chunk for chunk in chunks) @pytest.mark.asyncio - async def test_send_splits_paragraphs_into_multiple_bubbles(self, monkeypatch): + async def test_send_preserves_paragraphs_in_one_bubble(self, monkeypatch): adapter = _make_adapter(monkeypatch) sent = [] @@ -102,7 +103,29 @@ async def fake_api_post(path, payload): result = await adapter.send("user@example.com", "first thought\n\nsecond thought") assert result.success is True - assert sent == ["first thought", "second thought"] + assert sent == ["first thought\n\nsecond thought"] + + @pytest.mark.asyncio + async def test_send_chunks_only_when_text_exceeds_limit(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + monkeypatch.setattr(type(adapter), "MAX_MESSAGE_LENGTH", 6) + sent = [] + + async def fake_resolve_chat_guid(chat_id): + return "iMessage;-;user@example.com" + + async def fake_api_post(path, payload): + sent.append(payload["message"]) + return {"data": {"guid": f"msg-{len(sent)}"}} + + monkeypatch.setattr(adapter, "_resolve_chat_guid", fake_resolve_chat_guid) + monkeypatch.setattr(adapter, "_api_post", fake_api_post) + + result = await adapter.send("user@example.com", "abcdefghij") + + assert result.success is True + assert len(sent) > 1 + assert "".join(sent) == "abcdefghij" def test_format_message_strips_markdown(self, monkeypatch): adapter = _make_adapter(monkeypatch) @@ -163,6 +186,116 @@ def test_clean_mention_text_strips_leading_wake_word(self, monkeypatch): assert adapter._clean_mention_text("Hermes agent: summarize this") == "summarize this" assert adapter._clean_mention_text("please ask Hermes about this") == "please ask Hermes about this" + def test_webhook_events_default_to_new_message_only(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + assert adapter.webhook_events == ["new-message"] + + def test_webhook_events_can_opt_in_to_updated_message(self, monkeypatch): + adapter = _make_adapter( + monkeypatch, + webhook_events=["new-message", "updated-message"], + ) + assert adapter.webhook_events == ["new-message", "updated-message"] + + def test_webhook_events_env_parses_comma_list(self, monkeypatch): + monkeypatch.setenv("BLUEBUBBLES_WEBHOOK_EVENTS", "new-message,updated-message") + adapter = _make_adapter(monkeypatch) + assert adapter.webhook_events == ["new-message", "updated-message"] + + def test_dedup_key_prefers_message_guid_and_text_hash(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + payload = {"type": "new-message", "data": {"guid": "msg-1", "text": "hello"}} + record = adapter._extract_payload_record(payload) or {} + first = adapter._message_dedup_key(payload, record, "hello") + second = adapter._message_dedup_key(payload, record, "hello again") + assert first.startswith("guid:msg-1:") + assert first != second + + @pytest.mark.asyncio + async def test_webhook_listener_port_conflict_enters_outbound_only_mode(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + cleaned = [] + + class Router: + def add_get(self, *args, **kwargs): + pass + + def add_post(self, *args, **kwargs): + pass + + class App: + def __init__(self): + self.router = Router() + + class Runner: + def __init__(self, app, **kwargs): + self.app = app + + async def setup(self): + pass + + async def cleanup(self): + cleaned.append(True) + + class Site: + def __init__(self, runner, host, port): + pass + + async def start(self): + raise OSError(98, "address already in use") + + class Web: + Application = App + AppRunner = Runner + TCPSite = Site + + @staticmethod + def Response(text=""): + return text + + started = await adapter._start_webhook_listener(Web) + + assert started is False + assert adapter._runner is None + assert cleaned == [True] + + @pytest.mark.asyncio + async def test_connect_skips_webhook_registration_when_listener_is_busy(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + registered = [] + unregistered = [] + + async def fake_api_get(path): + if path == "/api/v1/server/info": + return {"data": {"private_api": True, "helper_connected": True}} + return {"status": 200} + + async def fake_start_listener(web): + return False + + async def fake_register(): + registered.append(True) + return True + + async def fake_unregister(): + unregistered.append(True) + return True + + monkeypatch.setattr(adapter, "_api_get", fake_api_get) + monkeypatch.setattr(adapter, "_start_webhook_listener", fake_start_listener) + monkeypatch.setattr(adapter, "_register_webhook", fake_register) + monkeypatch.setattr(adapter, "_unregister_webhook", fake_unregister) + + ok = await adapter.connect() + try: + assert ok is True + assert adapter.is_connected is True + assert adapter._owns_webhook_listener is False + assert registered == [] + finally: + await adapter.disconnect() + assert unregistered == [] + class _FakeBlueBubblesRequest: def __init__(self, payload, password="secret"): @@ -193,7 +326,7 @@ async def fake_handle_message(event): "data": { "guid": "msg-1", "text": "casual family chatter", - "handle": {"address": "+15555550100"}, + "handle": {"address": "+155****0100"}, "isFromMe": False, "isGroup": True, "chats": [{"guid": "iMessage;+;group-chat"}], @@ -222,7 +355,7 @@ async def fake_handle_message(event): "data": { "guid": "msg-2", "text": "Hermes, summarize this", - "handle": {"address": "+15555550100"}, + "handle": {"address": "+155****0100"}, "isFromMe": False, "isGroup": True, "chats": [{"guid": "iMessage;+;group-chat"}], @@ -263,6 +396,129 @@ async def fake_handle_message(event): assert [event.text for event in handled] == ["hello from a dm"] +class TestBlueBubblesWebhookHandling: + @pytest.mark.asyncio + async def test_updated_message_status_only_is_acknowledged_but_not_processed(self, monkeypatch): + adapter = _make_adapter(monkeypatch) + handled = [] + + async def fake_handle_message(event): + handled.append(event) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + payload = { + "type": "updated-message", + "data": { + "guid": "msg-1", + "text": "hello", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + "chats": [{"guid": "iMessage;-;user@example.com"}], + }, + } + + response = await adapter._handle_webhook(_FakeBlueBubblesRequest(payload)) + await asyncio.sleep(0) + + assert response.text == "ok" + assert handled == [] + + @pytest.mark.asyncio + async def test_updated_message_edit_is_processed_once_when_opted_in(self, monkeypatch): + adapter = _make_adapter(monkeypatch, webhook_events=["new-message", "updated-message"]) + handled = [] + + async def fake_handle_message(event): + handled.append(event.text) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + payload = { + "type": "updated-message", + "data": { + "guid": "msg-edit-1", + "text": "hello edited", + "dateEdited": 123, + "handle": {"address": "user@example.com"}, + "isFromMe": False, + "chats": [{"guid": "iMessage;-;user@example.com"}], + }, + } + + response = await adapter._handle_webhook(_FakeBlueBubblesRequest(payload)) + await asyncio.sleep(0) + + assert response.text == "ok" + assert handled == ["hello edited"] + + @pytest.mark.asyncio + async def test_duplicate_guid_and_text_only_processes_once(self, monkeypatch): + adapter = _make_adapter(monkeypatch, webhook_events=["new-message", "updated-message"]) + handled = [] + + async def fake_handle_message(event): + handled.append(event.text) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + payload = { + "type": "new-message", + "data": { + "guid": "msg-2", + "text": "hello", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + "chats": [{"guid": "iMessage;-;user@example.com"}], + }, + } + + first = await adapter._handle_webhook(_FakeBlueBubblesRequest(payload)) + second = await adapter._handle_webhook(_FakeBlueBubblesRequest(payload)) + await asyncio.sleep(0) + + assert first.text == "ok" + assert second.text == "ok" + assert handled == ["hello"] + + @pytest.mark.asyncio + async def test_duplicate_dm_guid_variants_only_process_once(self, monkeypatch): + import asyncio + + adapter = _make_adapter(monkeypatch, webhook_events=["new-message", "updated-message"]) + handled = [] + + async def fake_handle_message(event): + handled.append((event.text, event.source.chat_id)) + + monkeypatch.setattr(adapter, "handle_message", fake_handle_message) + first_payload = { + "type": "new-message", + "data": { + "guid": "msg-guid-a", + "text": "same text", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + "chats": [{"guid": "any;-;user@example.com"}], + }, + } + second_payload = { + "type": "new-message", + "data": { + "guid": "msg-guid-b", + "text": "same text", + "handle": {"address": "user@example.com"}, + "isFromMe": False, + "chatIdentifier": "user@example.com", + }, + } + + first = await adapter._handle_webhook(_FakeBlueBubblesRequest(first_payload)) + second = await adapter._handle_webhook(_FakeBlueBubblesRequest(second_payload)) + await asyncio.sleep(0) + + assert first.text == "ok" + assert second.text == "ok" + assert handled == [("same text", "user@example.com")] + + class TestBlueBubblesWebhookParsing: def test_webhook_prefers_chat_guid_over_message_guid(self, monkeypatch): adapter = _make_adapter(monkeypatch)