From 4192eab81f01ba075782bf2860ff9d4fcc315a38 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Wed, 10 Jun 2026 12:20:50 -0500 Subject: [PATCH] fix: stabilize BlueBubbles webhook handling --- gateway/platforms/bluebubbles.py | 59 ++++++++++++++++--- run_agent.py | 3 + tests/gateway/test_bluebubbles.py | 40 +++++++++++++ .../test_auxiliary_failure_visibility.py | 22 +++++++ tools/send_message_tool.py | 2 +- 5 files changed, 117 insertions(+), 9 deletions(-) create mode 100644 tests/run_agent/test_auxiliary_failure_visibility.py diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py index c2213daeef1e7..8d4bf4e4e0aff 100644 --- a/gateway/platforms/bluebubbles.py +++ b/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"), diff --git a/run_agent.py b/run_agent.py index 5465bb9ae2de2..ae1d0fe52864e 100644 --- a/run_agent.py +++ b/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]: diff --git a/tests/gateway/test_bluebubbles.py b/tests/gateway/test_bluebubbles.py index 7d4a71378c0bf..a4adbca13dfe7 100644 --- a/tests/gateway/test_bluebubbles.py +++ b/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 diff --git a/tests/run_agent/test_auxiliary_failure_visibility.py b/tests/run_agent/test_auxiliary_failure_visibility.py new file mode 100644 index 0000000000000..2f139ddda8c79 --- /dev/null +++ b/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:") diff --git a/tools/send_message_tool.py b/tools/send_message_tool.py index afa473e384bcc..5fa6ea2f59229 100644 --- a/tools/send_message_tool.py +++ b/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: