""" 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 3. Health stats tracking — poll stats, reconnect counts, error rates 4. Dynamic @all-bots resolution — falls back to hardcoded default Architecture (mirrors pi-zulip-extension): Zulip event queue -> poll loop -> parse message -> handle_message(MessageEvent) -> Gateway processes -> send() / edit_message() posts response to Zulip No external SDK required — only httpx (already a Hermes dependency). Ships as a Hermes platform plugin under plugins/platforms/zulip/. """ import asyncio import base64 import json import logging import os import re import time import uuid from collections import deque from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple try: import httpx HTTPX_AVAILABLE = True except ImportError: HTTPX_AVAILABLE = False httpx = None from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, MessageType, SendResult, ) logger = logging.getLogger(__name__) # Constants DEFAULT_STREAM = "agent-hub" DEFAULT_ALL_BOTS_USER_ID = 1 DEFAULT_POLL_INTERVAL = 3.0 MAX_ZULIP_MESSAGE = 10000 ECHO_TAG_PREFIX = "hermes-zulip-" 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 MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*") # Placeholder messages for streaming UX PLACEHOLDERS = [ ":robot: _Processing your message..._", ":hourglass_flowing_sand: _Thinking..._", ":brain: _Generating response..._", ] def _build_auth_header(email: str, api_key: str) -> str: """Build Basic auth header for Zulip API.""" token = base64.b64encode(f"{email}:{api_key}".encode()).decode() return f"Basic {token}" def _truncate(text: str, limit: int = MAX_ZULIP_MESSAGE) -> str: """Truncate to Zulip's message limit with notice.""" if len(text) <= limit: return text return text[:limit] + "\n\n[...truncated at Zulip limit]" def _parse_zulip_timestamp(ts: Any) -> datetime: """Convert Zulip timestamp (float seconds) to datetime.""" try: return datetime.fromtimestamp(float(ts), tz=timezone.utc) except (ValueError, OSError, TypeError): return datetime.now(tz=timezone.utc) class ZulipAdapter(BasePlatformAdapter): """Zulip adapter for Hermes agents. Connects to Zulip via event queues, polls for new events, converts 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) - Health stats for monitoring - Dynamic @all-bots resolution """ def __init__(self, config: PlatformConfig): platform = Platform("zulip") super().__init__(config=config, platform=platform) extra = config.extra or {} # Zulip connection self._site: str = ( extra.get("site") or os.getenv("ZULIP_SITE", "") ).rstrip("/") self._email: str = extra.get("email") or os.getenv("ZULIP_EMAIL", "") self._api_key: str = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "") # Agent identity self._agent_name: str = ( extra.get("agent_name") or os.getenv("ZULIP_AGENT_NAME", "hermes-agent") ) # Stream config self._stream: str = ( extra.get("stream") or os.getenv("ZULIP_STREAM", DEFAULT_STREAM) ) # Self-test config — owner user ID for DM self-tests self._owner_user_id: Optional[int] = self._resolve_int_opt( extra.get("owner_user_id") or os.getenv("ZULIP_OWNER_USER_ID", "") ) # @all-bots user ID — dynamically resolved on connect, falls back to config/env/default self._all_bots_user_id: int = self._resolve_int( extra.get("all_bots_user_id") or os.getenv("ZULIP_ALL_BOTS_USER_ID", str(DEFAULT_ALL_BOTS_USER_ID)), DEFAULT_ALL_BOTS_USER_ID, ) # Polling self._poll_interval: float = float( extra.get("poll_interval") or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL)) ) # --- State --- self._auth_header: str = _build_auth_header(self._email, self._api_key) self._queue_id: Optional[str] = None self._last_event_id: int = -1 self._poll_task: Optional[asyncio.Task] = None self._dedup_cleanup_task: Optional[asyncio.Task] = None self._http_client: Optional[httpx.AsyncClient] = None self._pending_replies: Dict[str, int] = {} # chat_id -> placeholder zulip_msg_id # Derived identity for echo-loop prevention self._bot_user_id: Optional[int] = None self._bot_email: str = self._email # --- Gen 2: Dedup (background-maintained) --- self._seen_message_ids: Dict[str, float] = {} # --- 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, "last_error_msg": None, } # --- 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.""" if val is None or val == "": return default try: return int(val) except (ValueError, TypeError): return default @staticmethod def _resolve_int_opt(val: Any) -> Optional[int]: """Resolve an optional integer config value.""" if val is None or val == "": return None try: return int(val) except (ValueError, TypeError): return None # ------------------------------------------------------------------ # Connection lifecycle # ------------------------------------------------------------------ async def connect(self) -> bool: """Connect to Zulip and register an event queue.""" if not HTTPX_AVAILABLE: logger.warning("[%s] httpx not installed", self.name) return False if not self._site or not self._email or not self._api_key: logger.warning("[%s] Zulip credentials not configured", self.name) return False try: self._http_client = httpx.AsyncClient(timeout=30.0) self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat() # Register event queue queue_resp = await self._api_call( "POST", "/api/v1/register", data={ "event_types": '["message"]', "apply_markdown": "true", "include_subscribers": "false", }, ) if not queue_resp: logger.error("[%s] Failed to register event queue", self.name) return False data = queue_resp self._queue_id = data.get("queue_id") self._last_event_id = data.get("last_event_id", -1) self._bot_user_id = data.get("user_id") # Gen 2: Try to resolve @all-bots user ID dynamically await self._resolve_all_bots_user_id() self._mark_connected() logger.info( "[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)", self.name, self._site, self._email, self._queue_id, self._bot_user_id, self._all_bots_user_id, ) # Start poll loop self._poll_task = asyncio.create_task(self._poll_forever()) # Gen 2: Start background dedup cleanup task self._dedup_cleanup_task = asyncio.create_task( 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: logger.error("[%s] Connection failed: %s", self.name, e) return False async def disconnect(self) -> None: """Disconnect from Zulip and cancel all background tasks.""" # Cancel dedup cleanup task if self._dedup_cleanup_task: self._dedup_cleanup_task.cancel() try: await self._dedup_cleanup_task except asyncio.CancelledError: 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() try: await self._poll_task except asyncio.CancelledError: pass self._poll_task = None self._queue_id = None self._mark_disconnected() logger.info("[%s] Disconnected", self.name) # ------------------------------------------------------------------ # Poll loop — receives messages from Zulip # ------------------------------------------------------------------ async def _poll_forever(self) -> None: """Continuously poll the Zulip event queue.""" backoff_idx = 0 while self._running: if not self._queue_id: await asyncio.sleep(self._poll_interval) continue try: events = await self._fetch_events() self._health_stats["poll_count"] += 1 self._health_stats["last_poll_at"] = datetime.now( timezone.utc ).isoformat() for event in events: 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 except Exception as e: if not self._running: break self._health_stats["poll_errors"] += 1 self._health_stats["last_error_at"] = datetime.now( timezone.utc ).isoformat() self._health_stats["last_error_msg"] = str(e)[:200] err_str = str(e) if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower(): logger.info("[%s] Queue expired, re-registering...", self.name) await self._reconnect() backoff_idx = 0 continue logger.warning("[%s] Poll error: %s", self.name, e) delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF) - 1)] backoff_idx += 1 await asyncio.sleep(delay) await asyncio.sleep(self._poll_interval) async def _fetch_events(self) -> List[Dict[str, Any]]: """Fetch events from the Zulip event queue.""" if not self._queue_id: return [] resp = await self._api_call( "GET", "/api/v1/events", params={ "queue_id": self._queue_id, "last_event_id": str(self._last_event_id), "dont_block": "true", }, ) if not resp: return [] events = resp.get("events", []) # Track last event ID for cursor advancement for event in events: eid = event.get("id", 0) if eid > self._last_event_id: self._last_event_id = eid return [e for e in events if e.get("type") == "message"] async def _reconnect(self) -> None: """Re-register the event queue.""" self._queue_id = None try: resp = await self._api_call( "POST", "/api/v1/register", data={ "event_types": '["message"]', "apply_markdown": "true", "include_subscribers": "false", }, ) if resp: self._queue_id = resp.get("queue_id") self._last_event_id = resp.get("last_event_id", -1) self._health_stats["reconnects"] += 1 logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id) except Exception as e: logger.error("[%s] Reconnect failed: %s", self.name, e) # ------------------------------------------------------------------ # Gen 2: Background dedup maintenance # ------------------------------------------------------------------ async def _dedup_cleanup_forever(self) -> None: """Periodically purge stale entries from the dedup map. 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: logger.debug( "[%s] Dedup cleanup error (non-fatal)", self.name ) def _prune_dedup_map(self) -> None: """Remove entries older than DEDUP_WINDOW from the dedup map.""" now = time.time() cutoff = now - DEDUP_WINDOW before = len(self._seen_message_ids) self._seen_message_ids = { k: v for k, v in self._seen_message_ids.items() if v > cutoff } after = len(self._seen_message_ids) if before != after: logger.debug( "[%s] Dedup map pruned: %d -> %d entries", 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 # ------------------------------------------------------------------ async def _resolve_all_bots_user_id(self) -> None: """Try to resolve the @all-bots user ID from the Zulip server. Falls back to the configured or default value if resolution fails. """ try: resp = await self._api_call("GET", "/api/v1/users") if not resp: return members = resp.get("members", []) for member in members: email = member.get("email", "") # @all-bots typically has an email pattern like all-bots@... if "all-bots" in email.lower(): 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, ) return except Exception as e: logger.debug( "[%s] Could not resolve @all-bots ID dynamically: %s. " "Using configured default %s.", self.name, e, self._all_bots_user_id, ) # ------------------------------------------------------------------ # Message processing — Zulip event -> MessageEvent -> Gateway # ------------------------------------------------------------------ async def _process_zulip_event(self, event: Dict[str, Any]) -> None: """Convert a Zulip message event to MessageEvent and dispatch to gateway.""" msg = event.get("message", {}) msg_type = msg.get("type", "") # "private" or "stream" sender_email = msg.get("sender_email", "") sender_name = msg.get("sender_full_name", "Unknown") sender_id = msg.get("sender_id") content = msg.get("content", "") # Echo-loop prevention: skip own messages if sender_email == self._bot_email: return # Gen 2: Check for self-test response if self._selftest_future and not self._selftest_future.done(): # Detect if this is a response to our self-test self._selftest_future.set_result(True) return # Deduplication (O(1) — no cleanup on this path) msg_id = f"{event.get('id', '')}:{msg.get('id', '')}" if self._is_duplicate(msg_id): return # Update health stats self._health_stats["messages_received"] += 1 self._health_stats["last_message_at"] = datetime.now(timezone.utc).isoformat() # Determine if this message targets this bot mentioned_users = msg.get("mentioned_users", []) or [] mentioned_ids = [u.get("user_id") for u in mentioned_users if isinstance(u, dict)] is_dm = msg_type == "private" is_direct_mention = self._bot_user_id and self._bot_user_id in mentioned_ids is_all_bots = self._all_bots_user_id in mentioned_ids # DM-first: process all private messages if is_dm: self._health_stats["dms_routed"] += 1 logger.info("[%s] DM from %s: %.60s", self.name, sender_name, content) await self._route_message( text=content, chat_id=str(sender_id), chat_name=sender_name, user_id=str(sender_id), user_name=sender_name, chat_type="dm", msg_id=msg_id, timestamp=msg.get("timestamp"), raw=event, ) return # Stream: only respond to @mentions and @all-bots if msg_type == "stream" and (is_direct_mention or is_all_bots): self._health_stats["mentions_routed"] += 1 stream_name = msg.get("display_recipient", "unknown") topic = msg.get("subject", "general") logger.info( "[%s] %s in #%s > %s: %.60s", self.name, "@mention" if is_direct_mention else "@all-bots", stream_name, topic, content, ) # Clean @mention artifacts from content clean_content = MENTION_CLEANER.sub("", content).strip() chat_id = f"{stream_name}:{topic}" await self._route_message( text=clean_content, chat_id=chat_id, chat_name=f"#{stream_name} > {topic}", user_id=str(sender_id), user_name=sender_name, chat_type="stream", thread_id=topic, chat_topic=topic, msg_id=msg_id, timestamp=msg.get("timestamp"), raw=event, reply_to_id=msg.get("id"), reply_to_type="stream", reply_to_stream=stream_name, reply_to_topic=topic, ) async def _route_message( self, text: str, chat_id: str, chat_name: str, user_id: str, user_name: str, chat_type: str, msg_id: str, timestamp: Any, raw: Dict[str, Any], thread_id: Optional[str] = None, chat_topic: Optional[str] = None, reply_to_id: Optional[int] = None, reply_to_type: Optional[str] = None, reply_to_stream: Optional[str] = None, reply_to_topic: Optional[str] = None, ) -> None: """Build MessageEvent and send to gateway via handle_message.""" source = self.build_source( chat_id=chat_id, chat_name=chat_name, chat_type=chat_type, user_id=user_id, user_name=user_name, thread_id=thread_id, chat_topic=chat_topic, chat_id_alt=str(reply_to_id) if reply_to_id else None, ) # Derive message type message_type = MessageType.TEXT message_event = MessageEvent( text=text, message_type=message_type, source=source, message_id=msg_id, raw_message=raw, timestamp=_parse_zulip_timestamp(timestamp), ) # Store reply metadata so send() can respond to the right thread if reply_to_id: self._pending_replies[chat_id] = reply_to_id logger.debug("[%s] Dispatching message from %s", self.name, user_name) await self.handle_message(message_event) # ------------------------------------------------------------------ # Outbound messaging — send responses to Zulip # ------------------------------------------------------------------ async def send( self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send a message to a Zulip chat. For DMs, chat_id is the user_id. For streams, format is "stream_name:topic". Sends a placeholder first if this is the first message in a turn, then edits it on subsequent calls. """ # Determine target type and IDs is_dm = not chat_id or ":" not in chat_id truncated = _truncate(content) if is_dm: # DM: chat_id is the user_id to = [int(chat_id)] payload = { "type": "private", "to": json.dumps(to), "content": truncated, } else: # Stream: chat_id is "stream_name:topic" parts = chat_id.split(":", 1) stream_name = parts[0] topic = parts[1] if len(parts) > 1 else "general" payload = { "type": "stream", "to": stream_name, "subject": topic, "content": truncated, } # Check if we should send a placeholder first (first message in turn) placeholder_msg_id = self._pending_replies.pop(chat_id, None) if placeholder_msg_id: # This is the first response — send placeholder and return its ID placeholder = self._pick_placeholder() placehold_payload = {**payload, "content": placeholder} result = await self._send_api_call(placehold_payload) if result and result.get("id"): msg_id = str(result["id"]) logger.debug( "[%s] Sent placeholder msg=%s for %s", self.name, msg_id, chat_id, ) self._health_stats["send_count"] += 1 return SendResult( success=True, platform="zulip", chat_id=chat_id, message_id=msg_id, metadata={"placeholder": True}, ) # Send actual content result = await self._send_api_call(payload) if result and result.get("id"): self._health_stats["send_count"] += 1 return SendResult( success=True, platform="zulip", chat_id=chat_id, message_id=str(result["id"]), ) self._health_stats["send_errors"] += 1 return SendResult( success=False, platform="zulip", chat_id=chat_id, error="Failed to send message", ) async def edit_message( self, chat_id: str, message_id: str, content: str, metadata: Optional[Dict[str, Any]] = None, ) -> bool: """Edit a previously sent Zulip message (for placeholder replacement).""" truncated = _truncate(content) payload = { "message_id": int(message_id), "content": truncated, } try: resp = await self._api_call("PATCH", "/api/v1/messages", data=payload) return resp is not None except Exception as e: logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e) return False async def send_typing(self, chat_id: str, metadata: Optional[Dict] = None) -> None: """Send typing indicator to a Zulip chat.""" if ":" in chat_id: return # No typing for stream messages try: user_ids = json.dumps([int(chat_id)]) await self._api_call( "POST", "/api/v1/typing", data={"to": user_ids, "op": "start"}, ) except Exception: pass # Non-critical async def stop_typing(self, chat_id: str) -> None: """Stop typing indicator.""" if ":" in chat_id: return try: user_ids = json.dumps([int(chat_id)]) await self._api_call( "POST", "/api/v1/typing", data={"to": user_ids, "op": "stop"}, ) except Exception: pass # Non-critical async def get_chat_info(self, chat_id: str) -> dict: """Return basic info about a Zulip chat.""" if ":" in chat_id: parts = chat_id.split(":", 1) return {"name": parts[0], "type": "channel", "topic": parts[1]} return {"name": chat_id, "type": "dm"} async def delete_message( self, chat_id: str, message_id: str, metadata: Optional[Dict] = None ) -> bool: """Delete a message via Zulip API.""" # Zulip doesn't support message deletion via API for bots # Send an empty edit instead return await self.edit_message(chat_id, message_id, "*deleted*") # ------------------------------------------------------------------ # Gen 2: Self-test diagnostics # ------------------------------------------------------------------ async def selftest(self) -> Dict[str, Any]: """Run a self-test to verify adapter health. Checks: 1. Connection state (queue registered, HTTP client alive) 2. Bot identity resolved (bot_user_id is set) 3. Queue polling active (poll_task is running) 4. Echo-loop prevention configured (bot_email matches registered email) 5. Subscription to primary stream (if reachable) 6. @all-bots ID resolved 7. Dedup maintenance task running Returns a dict with pass/fail for each check and an overall verdict. """ checks = {} # 1. Connection state checks["connected"] = { "status": self._connected, "detail": "Queue registered and connected" if self._connected else "Not connected to Zulip", } # 2. Queue registered checks["queue_registered"] = { "status": self._queue_id is not None, "detail": f"Queue: {self._queue_id}" if self._queue_id else "No queue registered", } # 3. HTTP client alive checks["http_client"] = { "status": self._http_client is not None and not self._http_client.is_closed, "detail": "HTTP client active" if self._http_client and not self._http_client.is_closed else "HTTP client unavailable", } # 4. Bot identity checks["bot_identity"] = { "status": self._bot_user_id is not None, "detail": f"Bot user_id: {self._bot_user_id}" if self._bot_user_id else "Bot user_id not resolved", } # 5. Poll loop active checks["poll_loop"] = { "status": self._poll_task is not None and not self._poll_task.done(), "detail": "Poll loop running" if self._poll_task and not self._poll_task.done() else "Poll loop not active", } # 6. Dedup cleanup active checks["dedup_cleanup"] = { "status": self._dedup_cleanup_task is not None and not self._dedup_cleanup_task.done(), "detail": "Dedup maintenance running" if self._dedup_cleanup_task and not self._dedup_cleanup_task.done() else "Dedup maintenance not active", } # 7. Echo-loop prevention checks["echo_prevention"] = { "status": bool(self._bot_email), "detail": f"Bot email: {self._bot_email}", } # 8. @all-bots configured checks["all_bots_configured"] = { "status": self._all_bots_user_id is not None, "detail": f"@all-bots user_id: {self._all_bots_user_id}", } # Overall verdict critical = ["connected", "queue_registered", "http_client", "poll_loop"] passed = sum(1 for c in checks.values() if c["status"]) critical_passed = sum(1 for k in critical if checks.get(k, {}).get("status")) failed = [k for k, v in checks.items() if not v["status"]] if critical_passed == len(critical) and passed == len(checks): verdict = "healthy" elif critical_passed < len(critical): verdict = "critical_failure" else: verdict = "degraded" return { "verdict": verdict, "timestamp": datetime.now(timezone.utc).isoformat(), "checks": checks, "summary": { "total": len(checks), "passed": passed, "failed": len(failed), "failed_checks": failed, "critical_passed": critical_passed, "critical_total": len(critical), }, } # ------------------------------------------------------------------ # 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 # ------------------------------------------------------------------ async def get_health_stats(self) -> Dict[str, Any]: """Return health and performance statistics. Returns a snapshot of: - Uptime - Poll counts, errors, reconnects - Message counts (total, DM, mention) - Send counts and errors - Dedup map size - Timestamps of last activity Suitable for periodic logging to RA-H OS knowledge graph. """ now = datetime.now(timezone.utc) uptime = None if self._health_stats.get("started_at"): try: started = datetime.fromisoformat(self._health_stats["started_at"]) uptime_seconds = (now - started).total_seconds() uptime = f"{uptime_seconds:.0f}s" except (ValueError, TypeError): pass stats = dict(self._health_stats) stats["uptime"] = uptime stats["dedup_map_size"] = len(self._seen_message_ids) stats["queue_id"] = self._queue_id stats["bot_user_id"] = self._bot_user_id stats["all_bots_user_id"] = self._all_bots_user_id stats["connected"] = self._connected stats["checked_at"] = now.isoformat() # Compute error rate total_polls = stats.get("poll_count", 0) or 1 stats["error_rate"] = round( stats.get("poll_errors", 0) / total_polls, 4 ) return stats # ------------------------------------------------------------------ # Zulip API helpers # ------------------------------------------------------------------ async def _api_call( self, method: str, path: str, data: Optional[Dict] = None, params: Optional[Dict] = None, ) -> Optional[Dict]: """Make an API call to Zulip.""" if not self._http_client: return None url = f"{self._site}{path}" headers = { "Authorization": self._auth_header, } try: if method == "GET": response = await self._http_client.get( url, params=params, headers=headers, ) elif method == "POST": response = await self._http_client.post( url, data=data, headers=headers, ) elif method == "PATCH": response = await self._http_client.patch( url, data=data, headers=headers, ) else: return None if response.status_code >= 400: logger.warning( "[%s] API %s %s: %d %s", self.name, method, path, response.status_code, response.text[:200], ) return None return response.json() except httpx.TimeoutException: logger.debug("[%s] Timeout on %s %s", self.name, method, path) return None except Exception as e: logger.warning("[%s] API error %s %s: %s", self.name, method, path, e) return None async def _send_api_call(self, payload: Dict) -> Optional[Dict]: """Send a message to Zulip.""" return await self._api_call("POST", "/api/v1/messages", data=payload) # ------------------------------------------------------------------ # Utilities # ------------------------------------------------------------------ def _pick_placeholder(self) -> str: """Pick a random placeholder message for streaming UX.""" idx = hash(str(time.time())) % len(PLACEHOLDERS) return PLACEHOLDERS[idx] def _is_duplicate(self, msg_id: str) -> bool: """Deduplication using message IDs — O(1) lookup, no cleanup. Cleanup runs in a separate background task (_dedup_cleanup_forever) so this path stays fast. """ if msg_id in self._seen_message_ids: return True 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 # ------------------------------------------------------------------ def check_requirements() -> bool: """Check whether the Zulip adapter is installable.""" if not HTTPX_AVAILABLE: return False site = os.getenv("ZULIP_SITE", "").strip() email = os.getenv("ZULIP_EMAIL", "").strip() api_key = os.getenv("ZULIP_API_KEY", "").strip() return bool(site and email and api_key) def validate_config(config) -> bool: """Validate that the configured Zulip platform has credentials.""" extra = getattr(config, "extra", {}) or {} site = extra.get("site") or os.getenv("ZULIP_SITE", "") email = extra.get("email") or os.getenv("ZULIP_EMAIL", "") api_key = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "") return bool(site and email and api_key) def is_connected(config) -> bool: """Check whether Zulip is configured (env or config.yaml).""" return validate_config(config) def _env_enablement() -> Optional[dict]: """Seeds PlatformConfig.extra from env vars for env-only setups.""" site = os.getenv("ZULIP_SITE", "").strip() email = os.getenv("ZULIP_EMAIL", "").strip() api_key = os.getenv("ZULIP_API_KEY", "").strip() if not (site and email and api_key): return None result: Dict[str, Any] = { "site": site, "email": email, "api_key": api_key, } stream = os.getenv("ZULIP_STREAM", "").strip() if stream: result["stream"] = stream agent_name = os.getenv("ZULIP_AGENT_NAME", "").strip() if agent_name: result["agent_name"] = agent_name home = os.getenv("ZULIP_HOME_CHANNEL", "").strip() if home: result["home_channel"] = home return result def register(ctx) -> None: """Plugin entry point — called by the Hermes plugin system at startup.""" ctx.register_platform( name="zulip", label="Zulip", adapter_factory=lambda cfg: ZulipAdapter(cfg), check_fn=check_requirements, validate_config=validate_config, is_connected=is_connected, required_env=["ZULIP_SITE", "ZULIP_EMAIL", "ZULIP_API_KEY"], install_hint="pip install httpx # already a Hermes dependency", env_enablement_fn=_env_enablement, cron_deliver_env_var="ZULIP_HOME_CHANNEL", allowed_users_env="ZULIP_ALLOWED_USERS", allow_all_env="ZULIP_ALLOW_ALL_USERS", max_message_length=MAX_ZULIP_MESSAGE, emoji="💬", pii_safe=True, allow_update_command=True, platform_hint=( "You are communicating via Zulip. Format responses with " "Zulip-compatible Markdown. Keep responses concise and " "well-structured. DMs are private conversations; stream " "messages are visible to all subscribers." ), )