From 0acf1068bf649aba4c242b119f7cbb814989526f Mon Sep 17 00:00:00 2001 From: "Abiba (pi)" Date: Fri, 26 Jun 2026 22:31:16 +0000 Subject: [PATCH] =?UTF-8?q?feat(hermes):=20Zulip=20adapter=20Gen=203=20?= =?UTF-8?q?=E2=80=94=20malformed=20message=20resilience,=20periodic=20@all?= =?UTF-8?q?-bots=20refresh,=20health=20callback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gen 3 improvements from build-zulip-plugin contract run: 1. Malformed message resilience - Per-event try/except in poll loop — one bad event never kills adapter - malformed_events counter tracked in health stats - Fulfills Success Criterion: handles_malformed_messages == true 2. Periodic @all-bots refresh - _all_bots_refresh_forever() background task re-resolves every hour - Cancelled gracefully on disconnect() - all_bots_refreshes counter in health stats 3. Health stats callback mechanism - set_health_callback(callback, interval=600) — register external consumer - _report_health_if_callback() triggers on dedup cleanup cycle - Designed for Hermes Gateway to wire to RA-H OS knowledge graph logging - Cross-contract integration: get_all_bots_user_id() and get_bot_user_id() exposed for zulip-mention-reliability contract --- plugins/platforms/zulip/adapter.py | 134 ++++++++++++++++++++++++++++- 1 file changed, 131 insertions(+), 3 deletions(-) diff --git a/plugins/platforms/zulip/adapter.py b/plugins/platforms/zulip/adapter.py index 8315f97..9e93e8a 100644 --- a/plugins/platforms/zulip/adapter.py +++ b/plugins/platforms/zulip/adapter.py @@ -1,10 +1,15 @@ """ -Zulip platform adapter (Hermes plugin) — Gen 2. +Zulip platform adapter (Hermes plugin) — Gen 3. Connects a Hermes agent to Zulip via event queue polling. DMs and @mentions are routed to the agent via the Hermes Gateway's handle_message() interface. Replies use a placeholder->edit streaming pattern for UX feedback. +Gen 3 Improvements (2026-06-26): + 1. Malformed message resilience — try/except wraps each event, no crash on bad JSON + 2. Periodic @all-bots refresh — background task re-resolves user ID every hour + 3. Health stats callback — external consumers can register for periodic stats + Gen 2 Improvements (2026-06-26): 1. Background dedup maintenance — O(1) per-message path, periodic cleanup task 2. Self-test capability — selftest() method verifying DM/mention/queue health @@ -58,6 +63,7 @@ RECONNECT_BACKOFF = [2, 5, 10, 30, 60] DEDUP_WINDOW = 300 # 5 minutes DEDUP_MAX_SIZE = 1000 DEDUP_CLEANUP_INTERVAL = 120 # Clean old entries every 2 minutes +ALL_BOTS_REFRESH_INTERVAL = 3600 # Re-resolve @all-bots user ID every hour SELFTEST_TIMEOUT = 30 # seconds to wait for self-test response # Regex to strip Zulip @mention markup @@ -99,6 +105,11 @@ class ZulipAdapter(BasePlatformAdapter): DMs and @mentions to MessageEvent, and sends responses with streaming placeholder->edit UX. + Gen 3 adds: + - Malformed message resilience (per-event try/except) + - Periodic @all-bots refresh (hourly background task) + - Health stats callback for external consumers + Gen 2 adds: - Background dedup maintenance task (O(1) per-message) - Self-test diagnostics (DM/mention/queue health) @@ -163,17 +174,19 @@ class ZulipAdapter(BasePlatformAdapter): # --- Gen 2: Dedup (background-maintained) --- self._seen_message_ids: Dict[str, float] = {} - # --- Gen 2: Health stats --- + # --- Gen 2/3: Health stats --- self._health_stats: Dict[str, Any] = { "started_at": None, "poll_count": 0, "poll_errors": 0, + "malformed_events": 0, "messages_received": 0, "dms_routed": 0, "mentions_routed": 0, "reconnects": 0, "send_count": 0, "send_errors": 0, + "all_bots_refreshes": 0, "last_poll_at": None, "last_message_at": None, "last_error_at": None, @@ -183,6 +196,13 @@ class ZulipAdapter(BasePlatformAdapter): # --- Gen 2: Self-test future for async response --- self._selftest_future: Optional[asyncio.Future] = None + # --- Gen 3: Periodic @all-bots refresh --- + self._all_bots_refresh_task: Optional[asyncio.Task] = None + + # --- Gen 3: Health stats callback --- + self._health_callback = None + self._health_callback_interval: int = 600 # 10 min + @staticmethod def _resolve_int(val: Any, default: int) -> int: """Resolve an integer config value, falling back to default.""" @@ -257,6 +277,11 @@ class ZulipAdapter(BasePlatformAdapter): self._dedup_cleanup_forever() ) + # Gen 3: Start periodic @all-bots refresh task + self._all_bots_refresh_task = asyncio.create_task( + self._all_bots_refresh_forever() + ) + return True except Exception as e: @@ -274,6 +299,15 @@ class ZulipAdapter(BasePlatformAdapter): pass self._dedup_cleanup_task = None + # Cancel @all-bots refresh task + if self._all_bots_refresh_task: + self._all_bots_refresh_task.cancel() + try: + await self._all_bots_refresh_task + except asyncio.CancelledError: + pass + self._all_bots_refresh_task = None + # Cancel poll task if self._poll_task: self._poll_task.cancel() @@ -307,7 +341,20 @@ class ZulipAdapter(BasePlatformAdapter): timezone.utc ).isoformat() for event in events: - await self._process_zulip_event(event) + try: + await self._process_zulip_event(event) + except Exception as e: + # Gen 3: Malformed message resilience — catch per-event + # failures so one bad message never kills the poll loop + logger.warning( + "[%s] Malformed event skipped: %s. " + "Event id=%s", + self.name, e, + event.get("id", "unknown"), + ) + self._health_stats["malformed_events"] = ( + self._health_stats.get("malformed_events", 0) + 1 + ) backoff_idx = 0 except asyncio.CancelledError: break @@ -387,11 +434,24 @@ class ZulipAdapter(BasePlatformAdapter): This runs as a background task so the per-message _is_duplicate() path stays O(1) — no cleanup cost on every message. + + Also triggers the health stats callback on each cycle if one is + registered — natural tick for periodic monitoring. """ + cycle = 0 while self._running: try: await asyncio.sleep(DEDUP_CLEANUP_INTERVAL) self._prune_dedup_map() + cycle += 1 + # Report health every N cycles based on callback interval + if ( + self._health_callback + and cycle * DEDUP_CLEANUP_INTERVAL + >= self._health_callback_interval + ): + await self._report_health_if_callback() + cycle = 0 except asyncio.CancelledError: break except Exception: @@ -414,6 +474,28 @@ class ZulipAdapter(BasePlatformAdapter): self.name, before, after, ) + # ------------------------------------------------------------------ + # Gen 3: Periodic @all-bots refresh + # ------------------------------------------------------------------ + + async def _all_bots_refresh_forever(self) -> None: + """Periodically re-resolve the @all-bots user ID. + + Runs every ALL_BOTS_REFRESH_INTERVAL (1 hour by default). + Handles the case where the @all-bots user is created or + recreated after initial connection. + """ + while self._running: + try: + await asyncio.sleep(ALL_BOTS_REFRESH_INTERVAL) + await self._resolve_all_bots_user_id() + except asyncio.CancelledError: + break + except Exception: + logger.debug( + "[%s] @all-bots refresh error (non-fatal)", self.name + ) + # ------------------------------------------------------------------ # Gen 2: Dynamic @all-bots resolution # ------------------------------------------------------------------ @@ -435,6 +517,7 @@ class ZulipAdapter(BasePlatformAdapter): uid = member.get("user_id") if uid is not None: self._all_bots_user_id = int(uid) + self._health_stats["all_bots_refreshes"] += 1 logger.info( "[%s] Resolved @all-bots user_id=%s from %s", self.name, uid, email, @@ -838,6 +921,39 @@ class ZulipAdapter(BasePlatformAdapter): }, } + # ------------------------------------------------------------------ + # Gen 3: Health stats callback + # ------------------------------------------------------------------ + + def set_health_callback( + self, + callback, + interval: int = 600, + ) -> None: + """Register a callback for periodic health stats reporting. + + The callback is called with health stats dict every `interval` + seconds. Can be used by the Hermes Gateway to log health to + RA-H OS knowledge graph or external monitoring. + + Args: + callback: Async callable receiving health stats dict + interval: Seconds between calls (default 600 = 10 min) + """ + self._health_callback = callback + self._health_callback_interval = interval + + async def _report_health_if_callback(self) -> None: + """Call the health stats callback if one is registered.""" + if self._health_callback: + try: + stats = await self.get_health_stats() + await self._health_callback(stats) + except Exception as e: + logger.debug( + "[%s] Health callback error: %s", self.name, e + ) + # ------------------------------------------------------------------ # Gen 2: Health stats # ------------------------------------------------------------------ @@ -960,6 +1076,18 @@ class ZulipAdapter(BasePlatformAdapter): self._seen_message_ids[msg_id] = time.time() return False + def get_all_bots_user_id(self) -> int: + """Return the current @all-bots user ID. + + This is used by the zulip-mention-reliability contract for + cross-contract verification. + """ + return self._all_bots_user_id + + def get_bot_user_id(self) -> Optional[int]: + """Return the bot's own user ID for identity checks.""" + return self._bot_user_id + # ------------------------------------------------------------------ # Plugin registration helpers