SyslogSolution/zulip-platform-plugins
7
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hermes): Zulip adapter Gen 3 — malformed message resilience, periodic @all-bots refresh, health callback

Browse Source 
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
This commit is contained in:
Abiba (pi)
2026-06-26 22:31:16 +00:00
parent 1c8f48fd77
commit 0acf1068bf
1 changed files with 131 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/platforms/zulip/adapter.py
+131 -3
View File
@@ -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