fix: prevent BlueBubbles duplicate replies

gateway/platforms/base.py

@@ -3457,6 +3457,8 @@ async def _stop_typing_task() -> None:
             if not response:
                 logger.debug("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id)
             if response:
+                await _stop_typing_task()
+
                 # Capture [[as_document]] before extract_media strips it, so the
                 # dispatch partition below can route image-extension files
                 # through send_document instead of send_multiple_images. Used

gateway/platforms/bluebubbles.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import re
+import time
 import uuid
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -55,6 +56,8 @@
 
 # Webhook event types that carry user messages
 _MESSAGE_EVENTS = {"new-message", "message", "updated-message"}
+_INBOUND_MESSAGE_STATE_TTL_SECONDS = 300
+_INBOUND_MESSAGE_STATE_MAX_SIZE = 512
 
 # Log redaction patterns
 _PHONE_RE = re.compile(r"\+?\d{7,15}")
@@ -129,6 +132,8 @@ def __init__(self, config: PlatformConfig):
         self._private_api_enabled: Optional[bool] = None
         self._helper_connected: bool = False
         self._guid_cache: Dict[str, str] = {}
+        self._inbound_message_state: Dict[str, tuple[str, str, float]] = {}
+        self._typing_active_chats: set[str] = set()
 
     # ------------------------------------------------------------------
     # API helpers
@@ -249,8 +254,8 @@ async def _find_registered_webhooks(self, url: str) -> list:
             data = res.get("data")
             if isinstance(data, list):
                 return [wh for wh in data if wh.get("url") == url]
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning("[bluebubbles] failed to list registered webhooks: %s", exc)
         return []
 
     async def _register_webhook(self) -> bool:
@@ -590,6 +595,7 @@ async def send_animation(
     # ------------------------------------------------------------------
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
+        """Show iMessage typing while Hermes is actively processing a turn."""
         if not self._private_api_enabled or not self._helper_connected or not self.client:
             return
         try:
@@ -599,10 +605,15 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
                 await self.client.post(
                     self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5
                 )
-        except Exception:
-            pass
+                self._typing_active_chats.add(chat_id)
+        except Exception as exc:
+            logger.warning("[bluebubbles] failed to send typing indicator: %s", exc)
 
     async def stop_typing(self, chat_id: str) -> None:
+        """Clear iMessage typing when Hermes finishes or is interrupted."""
+        if chat_id not in self._typing_active_chats:
+            return
+        self._typing_active_chats.discard(chat_id)
         if not self._private_api_enabled or not self._helper_connected or not self.client:
             return
         try:
@@ -612,8 +623,8 @@ async def stop_typing(self, chat_id: str) -> None:
                 await self.client.delete(
                     self._api_url(f"/api/v1/chat/{encoded}/typing"), timeout=5
                 )
-        except Exception:
-            pass
+        except Exception as exc:
+            logger.warning("[bluebubbles] failed to stop typing indicator: %s", exc)
 
     # ------------------------------------------------------------------
     # Read receipts
@@ -765,6 +776,65 @@ def _value(*candidates: Any) -> Optional[str]:
                 return candidate.strip()
         return None
 
+    def _prune_inbound_message_state(self, now: float) -> None:
+        cutoff = now - _INBOUND_MESSAGE_STATE_TTL_SECONDS
+        expired = [
+            message_id
+            for message_id, (_text, _chat_id, ts) in self._inbound_message_state.items()
+            if ts < cutoff
+        ]
+        for message_id in expired:
+            self._inbound_message_state.pop(message_id, None)
+        while len(self._inbound_message_state) > _INBOUND_MESSAGE_STATE_MAX_SIZE:
+            self._inbound_message_state.pop(next(iter(self._inbound_message_state)))
+
+    def _classify_inbound_message_update(
+        self,
+        event_type: str,
+        message_id: Optional[str],
+        text: str,
+        chat_id: str,
+    ) -> tuple[str, str, Optional[str]]:
+        """Classify BlueBubbles message events before dispatch.
+
+        BlueBubbles ``updated-message`` means the macOS Messages DB row was
+        updated; it is not synonymous with a user edit.  Delivery/read status,
+        chat relationship hydration, attachment metadata, and real user edits
+        can all arrive through the same webhook type.  Hermes must therefore
+        compare message identity + text before creating a new user turn.
+
+        Returns ``(kind, canonical_chat_id, previous_text)`` where kind is:
+        - ``new``: dispatch normally
+        - ``metadata``: acknowledge only; same text for an already-seen message
+        - ``stale_update``: acknowledge only; update arrived without prior text
+        - ``edit``: dispatch an explicit correction turn
+        """
+        now = time.monotonic()
+        self._prune_inbound_message_state(now)
+        prior = self._inbound_message_state.get(message_id) if message_id else None
+        if event_type == "updated-message":
+            if not message_id:
+                return "stale_update", chat_id, None
+            if prior is None:
+                # BlueBubbles updates are row lifecycle events (read/delivered/edit/etc.),
+                # not new user turns.  If Hermes missed the original new-message
+                # (restart or cache expiry), we cannot distinguish a real edit
+                # from read/delivery metadata, so acknowledge and remember this
+                # snapshot instead of creating a duplicate turn.
+                self._inbound_message_state[message_id] = (text, chat_id, now)
+                return "stale_update", chat_id, None
+
+            previous_text, previous_chat_id, _previous_ts = prior
+            canonical_chat_id = previous_chat_id or chat_id
+            self._inbound_message_state[message_id] = (text, canonical_chat_id, now)
+            if previous_text == text:
+                return "metadata", canonical_chat_id, previous_text
+            return "edit", canonical_chat_id, previous_text
+
+        if message_id:
+            self._inbound_message_state[message_id] = (text, chat_id, now)
+        return "new", chat_id, None
+
     async def _handle_webhook(self, request):
         from aiohttp import web
 
@@ -898,7 +968,37 @@ 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
+        session_chat_id = str(chat_guid or chat_identifier)
+        message_id = self._value(
+            record.get("guid"),
+            record.get("messageGuid"),
+            record.get("id"),
+        )
+        update_kind, canonical_chat_id, previous_text = self._classify_inbound_message_update(
+            event_type,
+            message_id,
+            text,
+            session_chat_id,
+        )
+        logger.info(
+            "[bluebubbles] webhook message event type=%s kind=%s message=%s chat=%s chat_identifier=%s sender=%s",
+            event_type or "message",
+            update_kind,
+            _redact(message_id or ""),
+            _redact(canonical_chat_id or session_chat_id or ""),
+            _redact(chat_identifier or ""),
+            _redact(sender or ""),
+        )
+        if update_kind in {"metadata", "stale_update"}:
+            return web.Response(text="ok")
+        if update_kind == "edit":
+            session_chat_id = canonical_chat_id
+            text = (
+                "User edited a previous iMessage.\n"
+                f"Previous text: {previous_text or ''}\n"
+                f"Edited text: {text}"
+            )
+
         is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or ""))
         source = self.build_source(
             chat_id=session_chat_id,
@@ -913,11 +1013,7 @@ async def _handle_webhook(self, request):
             message_type=msg_type,
             source=source,
             raw_message=payload,
-            message_id=self._value(
-                record.get("guid"),
-                record.get("messageGuid"),
-                record.get("id"),
-            ),
+            message_id=message_id,
             reply_to_message_id=self._value(
                 record.get("threadOriginatorGuid"),
                 record.get("associatedMessageGuid"),

tests/gateway/test_bluebubbles.py

@@ -7,6 +7,9 @@
 def _make_adapter(monkeypatch, **extra):
     monkeypatch.setenv("BLUEBUBBLES_SERVER_URL", "http://localhost:1234")
     monkeypatch.setenv("BLUEBUBBLES_PASSWORD", "secret")
+    monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_HOST", raising=False)
+    monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_PORT", raising=False)
+    monkeypatch.delenv("BLUEBUBBLES_WEBHOOK_PATH", raising=False)
     from gateway.platforms.bluebubbles import BlueBubblesAdapter
 
     cfg = PlatformConfig(
@@ -132,6 +135,17 @@ def test_server_url_adds_scheme(self, monkeypatch):
 
 
 class TestBlueBubblesWebhookParsing:
+    class _Request:
+        def __init__(self, payload, password="secret"):
+            import json
+
+            self.query = {"password": password}
+            self.headers = {}
+            self._body = json.dumps(payload).encode("utf-8")
+
+        async def read(self):
+            return self._body
+
     def test_webhook_prefers_chat_guid_over_message_guid(self, monkeypatch):
         adapter = _make_adapter(monkeypatch)
         payload = {
@@ -273,6 +287,141 @@ def test_extract_payload_record_fallback_to_message(self, monkeypatch):
         record = adapter._extract_payload_record(payload)
         assert record["text"] == "hello"
 
+    @pytest.mark.asyncio
+    async def test_updated_message_metadata_update_does_not_create_second_turn(self, monkeypatch):
+        adapter = _make_adapter(monkeypatch)
+        handled = []
+
+        async def fake_handle_message(event):
+            handled.append(event)
+
+        monkeypatch.setattr(adapter, "handle_message", fake_handle_message)
+        base_record = {
+            "guid": "msg-guid-1",
+            "text": "hello",
+            "chatGuid": "iMessage;-;user@example.com",
+            "chatIdentifier": "user@example.com",
+            "handle": {"address": "user@example.com"},
+            "isFromMe": False,
+        }
+
+        first = await adapter._handle_webhook(
+            self._Request({"type": "new-message", "data": base_record})
+        )
+        second_record = dict(base_record)
+        second_record.pop("chatGuid")
+        second = await adapter._handle_webhook(
+            self._Request({"type": "updated-message", "data": second_record})
+        )
+
+        import asyncio
+
+        await asyncio.sleep(0)
+        assert first.status == 200
+        assert second.status == 200
+        assert len(handled) == 1
+        assert handled[0].source.chat_id == "iMessage;-;user@example.com"
+
+    @pytest.mark.asyncio
+    async def test_updated_message_without_prior_state_is_acknowledged_not_dispatched(self, monkeypatch):
+        adapter = _make_adapter(monkeypatch)
+        handled = []
+
+        async def fake_handle_message(event):
+            handled.append(event)
+
+        monkeypatch.setattr(adapter, "handle_message", fake_handle_message)
+        record = {
+            "guid": "msg-guid-1",
+            "text": "hello",
+            "chatIdentifier": "user@example.com",
+            "handle": {"address": "user@example.com"},
+            "isFromMe": False,
+        }
+
+        response = await adapter._handle_webhook(
+            self._Request({"type": "updated-message", "data": record})
+        )
+
+        import asyncio
+
+        await asyncio.sleep(0)
+        assert response.status == 200
+        assert len(handled) == 0
+
+    @pytest.mark.asyncio
+    async def test_updated_message_text_change_becomes_explicit_edit_turn(self, monkeypatch):
+        adapter = _make_adapter(monkeypatch)
+        handled = []
+
+        async def fake_handle_message(event):
+            handled.append(event)
+
+        monkeypatch.setattr(adapter, "handle_message", fake_handle_message)
+        base_record = {
+            "guid": "msg-guid-1",
+            "text": "helo",
+            "chatGuid": "iMessage;-;user@example.com",
+            "chatIdentifier": "user@example.com",
+            "handle": {"address": "user@example.com"},
+            "isFromMe": False,
+        }
+
+        await adapter._handle_webhook(
+            self._Request({"type": "new-message", "data": base_record})
+        )
+        edited_record = dict(base_record)
+        edited_record["text"] = "hello"
+        edited_record.pop("chatGuid")
+        response = await adapter._handle_webhook(
+            self._Request({"type": "updated-message", "data": edited_record})
+        )
+
+        import asyncio
+
+        await asyncio.sleep(0)
+        assert response.status == 200
+        assert len(handled) == 2
+        assert handled[1].source.chat_id == "iMessage;-;user@example.com"
+        assert handled[1].message_id == "msg-guid-1"
+        assert handled[1].text == (
+            "User edited a previous iMessage.\n"
+            "Previous text: helo\n"
+            "Edited text: hello"
+        )
+
+
+    @pytest.mark.asyncio
+    async def test_typing_endpoints_send_while_processing_and_clear_afterwards(self, monkeypatch):
+        adapter = _make_adapter(monkeypatch)
+        calls = []
+
+        class FakeClient:
+            async def post(self, url, timeout=None, **kwargs):
+                calls.append(("POST", url, timeout))
+
+            async def delete(self, url, timeout=None, **kwargs):
+                calls.append(("DELETE", url, timeout))
+
+        monkeypatch.setattr(adapter, "client", FakeClient())
+        adapter._private_api_enabled = True
+        adapter._helper_connected = True
+
+        async def fake_resolve(chat_id):
+            return "iMessage;-;user@example.com"
+
+        monkeypatch.setattr(adapter, "_resolve_chat_guid", fake_resolve)
+
+        await adapter.send_typing("iMessage;-;user@example.com")
+        await adapter.stop_typing("iMessage;-;user@example.com")
+        await adapter.stop_typing("iMessage;-;user@example.com")
+
+        assert len(calls) == 2
+        assert calls[0][0] == "POST"
+        assert "/api/v1/chat/iMessage%3B-%3Buser%40example.com/typing" in calls[0][1]
+        assert calls[1][0] == "DELETE"
+        assert "/api/v1/chat/iMessage%3B-%3Buser%40example.com/typing" in calls[1][1]
+
 
 class TestBlueBubblesGuidResolution:
     def test_raw_guid_returned_as_is(self, monkeypatch):