fix: stabilize BlueBubbles webhook handling

gateway/platforms/bluebubbles.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import re
+import time
 import uuid
 from collections import OrderedDict
 from datetime import datetime
@@ -43,6 +44,7 @@
 DEFAULT_WEBHOOK_PORT = 8645
 DEFAULT_WEBHOOK_PATH = "/bluebubbles-webhook"
 MAX_TEXT_LENGTH = 4000
+INBOUND_DEDUPE_TTL_SECS = 90.0
 
 # BlueBubbles/iMessage does not expose a stable bot mention identity like
 # Slack (<@U...>), Telegram (@botname), or Matrix (MXID). When users opt into
@@ -147,9 +149,11 @@ def __init__(self, config: PlatformConfig):
         )
         self.client: Optional[httpx.AsyncClient] = None
         self._runner = None
+        self._webhook_registered_by_instance = False
         self._private_api_enabled: Optional[bool] = None
         self._helper_connected: bool = False
         self._guid_cache: OrderedDict[str, str] = OrderedDict()
+        self._seen_inbound_events: OrderedDict[str, float] = OrderedDict()
 
     # ------------------------------------------------------------------
     # API helpers
@@ -200,6 +204,25 @@ def _message_matches_mention_patterns(self, text: str) -> bool:
             return False
         return any(pattern.search(text) for pattern in self._mention_patterns)
 
+    def _seen_inbound_event(self, key: str) -> bool:
+        """Return True if an inbound BlueBubbles webhook was already handled."""
+        if not key:
+            return False
+        now = time.monotonic()
+        # Opportunistically prune expired entries so the cache stays bounded.
+        while self._seen_inbound_events:
+            _, ts = next(iter(self._seen_inbound_events.items()))
+            if now - ts <= INBOUND_DEDUPE_TTL_SECS:
+                break
+            self._seen_inbound_events.popitem(last=False)
+        if key in self._seen_inbound_events:
+            self._seen_inbound_events.move_to_end(key)
+            return True
+        self._seen_inbound_events[key] = now
+        while len(self._seen_inbound_events) > 1000:
+            self._seen_inbound_events.popitem(last=False)
+        return False
+
     def _clean_mention_text(self, text: str) -> str:
         """Strip a leading BlueBubbles wake word before dispatch.
 
@@ -231,7 +254,7 @@ async def _api_post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
     # Lifecycle
     # ------------------------------------------------------------------
 
-    async def connect(self) -> bool:
+    async def connect(self, *, start_webhook: bool = True) -> bool:
         if not self.server_url or not self.password:
             logger.error(
                 "[bluebubbles] BLUEBUBBLES_SERVER_URL and BLUEBUBBLES_PASSWORD are required"
@@ -263,6 +286,11 @@ async def connect(self) -> bool:
                 self.client = None
             return False
 
+        if not start_webhook:
+            self._webhook_registered_by_instance = False
+            self._mark_connected()
+            return True
+
         app = web.Application()
         app.router.add_get("/health", lambda _: web.Response(text="ok"))
         app.router.add_post(self.webhook_path, self._handle_webhook)
@@ -283,13 +311,15 @@ async def connect(self) -> bool:
 
         # Register webhook with BlueBubbles server
         # This is required for the server to know where to send events
-        await self._register_webhook()
+        self._webhook_registered_by_instance = await self._register_webhook()
 
         return True
 
     async def disconnect(self) -> None:
         # Unregister webhook before cleaning up
-        await self._unregister_webhook()
+        if self._webhook_registered_by_instance:
+            await self._unregister_webhook()
+            self._webhook_registered_by_instance = False
 
         if self.client:
             await self.client.aclose()
@@ -993,6 +1023,23 @@ 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)
 
+        message_id = self._value(
+            record.get("guid"),
+            record.get("messageGuid"),
+            record.get("id"),
+        )
+        dedupe_keys: List[str] = []
+        if message_id:
+            dedupe_keys.append(f"id:{message_id}")
+        # BlueBubbles can deliver the same inbound event under both a raw chat
+        # GUID (for example any;-;+1...) and a normalized phone-number chat id.
+        # Use sender/text as a fallback so one user message cannot fork into two
+        # Hermes sessions and send stale/old replies twice.
+        dedupe_keys.append(f"fallback:{sender}:{text}")
+        if any(self._seen_inbound_event(key) for key in dedupe_keys):
+            logger.info("[bluebubbles] duplicate inbound message ignored")
+            return web.Response(text="ok")
+
         session_chat_id = chat_guid or chat_identifier
         is_group = bool(record.get("isGroup")) or (";+;" in (chat_guid or ""))
         if is_group and self.require_mention:
@@ -1015,11 +1062,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"),

run_agent.py

@@ -1018,6 +1018,9 @@ def _emit_auxiliary_failure(self, task: str, exc: BaseException) -> None:
         detail = (detail or exc.__class__.__name__).strip()
         if len(detail) > 220:
             detail = detail[:217].rstrip() + "..."
+        if task.strip().lower() == "title generation":
+            logger.debug("Auxiliary title generation failed: %s", detail)
+            return
         self._emit_warning(f"⚠ Auxiliary {task} failed: {detail}")
 
     def _current_main_runtime(self) -> Dict[str, str]:

tests/gateway/test_bluebubbles.py

@@ -830,3 +830,43 @@ async def bad_get(path):
             adapter._unregister_webhook()
         )
         assert ok is False
+
+    def test_connect_outbound_only_skips_webhook_lifecycle(self, monkeypatch):
+        import asyncio
+
+        adapter = _make_adapter(monkeypatch)
+        register_called = False
+        unregister_called = False
+
+        async def fake_get(path):
+            if path == "/api/v1/ping":
+                return {"status": 200}
+            if path == "/api/v1/server/info":
+                return {"data": {"private_api": True, "helper_connected": True}}
+            raise AssertionError(path)
+
+        async def fake_register():
+            nonlocal register_called
+            register_called = True
+            return True
+
+        async def fake_unregister():
+            nonlocal unregister_called
+            unregister_called = True
+            return True
+
+        adapter._api_get = fake_get
+        adapter._register_webhook = fake_register
+        adapter._unregister_webhook = fake_unregister
+
+        ok = asyncio.get_event_loop().run_until_complete(
+            adapter.connect(start_webhook=False)
+        )
+        assert ok is True
+        assert adapter.client is not None
+        assert adapter._runner is None
+        assert adapter._webhook_registered_by_instance is False
+        assert register_called is False
+
+        asyncio.get_event_loop().run_until_complete(adapter.disconnect())
+        assert unregister_called is False

tests/run_agent/test_auxiliary_failure_visibility.py

@@ -0,0 +1,22 @@
+from run_agent import AIAgent
+
+
+def _make_agent():
+    agent = AIAgent.__new__(AIAgent)
+    warnings = []
+    agent._emit_warning = warnings.append
+    agent._summarize_api_error = lambda exc: str(exc)
+    return agent, warnings
+
+
+def test_title_generation_aux_failure_not_user_visible():
+    agent, warnings = _make_agent()
+    agent._emit_auxiliary_failure("title generation", TypeError("NoneType object is not iterable"))
+    assert warnings == []
+
+
+def test_non_title_aux_failure_still_user_visible():
+    agent, warnings = _make_agent()
+    agent._emit_auxiliary_failure("vision", RuntimeError("provider timeout"))
+    assert len(warnings) == 1
+    assert warnings[0].startswith("⚠ Auxiliary vision failed:")

tools/send_message_tool.py

@@ -1580,7 +1580,7 @@ async def _send_bluebubbles(extra, chat_id, message):
         from gateway.config import PlatformConfig
         pconfig = PlatformConfig(extra=extra)
         adapter = BlueBubblesAdapter(pconfig)
-        connected = await adapter.connect()
+        connected = await adapter.connect(start_webhook=False)
         if not connected:
             return _error("BlueBubbles: failed to connect to server")
         try: