"""Zulip platform adapter (Hermes plugin). Connects to a Zulip server via the Zulip Python SDK, registers an event queue for real-time message delivery, and polls for incoming events. DM-first architecture: private messages route directly into the Hermes agent's session, maintaining conversation continuity with TUI/CLI. Configuration in config.yaml:: platforms: zulip: enabled: true extra: email: "tanko-bot@chat.sysloggh.net" api_key: "${ZULIP_API_KEY}" site: "https://chat.sysloggh.net" poll_interval_ms: 3000 Environment variables (env wins over config.yaml): ZULIP_EMAIL Bot email (required) ZULIP_API_KEY Bot API key (required) ZULIP_SITE Server URL (required) ZULIP_ALLOWED_USERS Comma-separated user emails allowed (optional) ZULIP_ALLOW_ALL_USERS Allow any user (optional) ZULIP_HOME_CHANNEL Default recipient for cron/notifications (optional) ZULIP_HOME_CHANNEL_NAME Human label for home channel (optional) """ import asyncio import json import logging import os import time import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) try: import zulip ZULIP_SDK_AVAILABLE = True except ImportError: ZULIP_SDK_AVAILABLE = False zulip = None # type: ignore[assignment] from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, MessageType, SendResult, ) # ── Constants ────────────────────────────────────────────────────────────── DEFAULT_POLL_INTERVAL_MS = 3000 MAX_MESSAGE_LENGTH = 10000 # Zulip max message body length RECONNECT_BACKOFF = [2, 5, 10, 30, 60] DEDUP_WINDOW_SECONDS = 300 DEDUP_MAX_SIZE = 1000 PLACEHOLDERS = [ ":robot: _Processing your message..._", ":hourglass_flowing_sand: _Thinking..._", ":brain: _Generating response..._", ] # ── Plugin registration helpers ──────────────────────────────────────────── def check_requirements() -> bool: """Check whether the Zulip SDK is available and minimally configured.""" if not ZULIP_SDK_AVAILABLE: return False email = os.getenv("ZULIP_EMAIL", "").strip() api_key = os.getenv("ZULIP_API_KEY", "").strip() site = os.getenv("ZULIP_SITE", "").strip() return bool(email and api_key and site) def validate_config(config) -> bool: """Validate Zulip platform has email and site configured.""" extra = getattr(config, "extra", {}) or {} email = extra.get("email") or os.getenv("ZULIP_EMAIL", "") site = extra.get("site") or os.getenv("ZULIP_SITE", "") return bool(email and site) def is_connected(config) -> bool: """Check whether Zulip is configured (env or config.yaml).""" extra = getattr(config, "extra", {}) or {} email = os.getenv("ZULIP_EMAIL") or extra.get("email", "") site = os.getenv("ZULIP_SITE") or extra.get("site", "") return bool(email and site) def _env_enablement() -> dict | None: """Seed PlatformConfig.extra from env vars during gateway config load. Returns None when Zulip isn't minimally configured. """ email = os.getenv("ZULIP_EMAIL", "").strip() api_key = os.getenv("ZULIP_API_KEY", "").strip() site = os.getenv("ZULIP_SITE", "").strip() if not (email and api_key and site): return None seed: dict = {"email": email, "site": site} home = os.getenv("ZULIP_HOME_CHANNEL", "").strip() if home: seed["home_channel"] = { "chat_id": home, "name": os.getenv("ZULIP_HOME_CHANNEL_NAME", home), } return seed async def _standalone_send( pconfig, chat_id: str, message: str, *, thread_id: Optional[str] = None, media_files: Optional[List[str]] = None, force_document: bool = False, ) -> Dict[str, Any]: """Out-of-process send for cron / send_message_tool fallbacks.""" if not ZULIP_SDK_AVAILABLE: return {"error": "zulip standalone send: zulip SDK not installed"} extra = getattr(pconfig, "extra", {}) or {} email = extra.get("email") or os.getenv("ZULIP_EMAIL", "") api_key = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "") site = extra.get("site") or os.getenv("ZULIP_SITE", "") if not (email and api_key and site): return {"error": "zulip standalone send: ZULIP_EMAIL, ZULIP_API_KEY, ZULIP_SITE required"} try: client = zulip.Client(email=email, api_key=api_key, site=site) result = client.send_message({ "type": "private", "to": chat_id, "content": message[:MAX_MESSAGE_LENGTH], }) if result.get("result") == "success": return { "success": True, "platform": "zulip", "chat_id": chat_id, "message_id": str(result.get("id", "")), } return {"error": f"zulip send failed: {result.get('msg', 'unknown')}"} except Exception as e: return {"error": f"zulip standalone send failed: {e}"} # ── Adapter ──────────────────────────────────────────────────────────────── class ZulipAdapter(BasePlatformAdapter): """Zulip platform adapter. Connects to Zulip, registers a real-time event queue, and polls for incoming messages. DMs are routed into the Hermes agent's session. """ MAX_MESSAGE_LENGTH = MAX_MESSAGE_LENGTH def __init__(self, config: PlatformConfig): platform = Platform("zulip") super().__init__(config=config, platform=platform) extra = config.extra or {} # Core Zulip config — env overrides config.yaml self._email: str = os.getenv("ZULIP_EMAIL") or extra.get("email", "") self._api_key: str = os.getenv("ZULIP_API_KEY") or extra.get("api_key", "") self._site: str = (os.getenv("ZULIP_SITE") or extra.get("site", "")).rstrip("/") # Polling config self._poll_interval: float = ( (extra.get("poll_interval_ms") or DEFAULT_POLL_INTERVAL_MS) / 1000.0 ) # State self._client: Optional["zulip.Client"] = None self._queue_id: Optional[str] = None self._last_event_id: int = -1 self._poll_task: Optional[asyncio.Task] = None # Message deduplication: event_id -> timestamp self._seen_messages: Dict[str, float] = {} # Pending replies for placeholder -> edit streaming. # Maps chat_id -> {placeholder_msg_id, ...} self._pending_replies: Dict[str, Dict[str, Any]] = {} # ── Connection lifecycle ──────────────────────────────────────────── async def connect(self) -> bool: """Connect to Zulip by registering an event queue.""" if not ZULIP_SDK_AVAILABLE: logger.warning( "[%s] zulip SDK not installed. Run: pip install zulip", self.name ) return False if not (self._email and self._api_key and self._site): logger.warning( "[%s] ZULIP_EMAIL, ZULIP_API_KEY, ZULIP_SITE not all configured", self.name, ) return False try: self._client = zulip.Client( email=self._email, api_key=self._api_key, site=self._site, client="Hermes-Zulip-Plugin/1.0.0", ) # Register event queue for message events queue_result = await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.register( event_types=["message"], fetch_event_types=["message"], ), ) self._queue_id = queue_result.get("queue_id") self._last_event_id = queue_result.get("last_event_id", -1) if not self._queue_id: logger.error( "[%s] Queue registration failed: %s", self.name, queue_result, ) return False self._poll_task = asyncio.create_task(self._poll_loop()) self._mark_connected() logger.info( "[%s] Connected — %s on %s, queue=%s last_event=%s", self.name, self._email, self._site, self._queue_id, self._last_event_id, ) return True except Exception as e: logger.error("[%s] Failed to connect: %s", self.name, e) return False async def disconnect(self) -> bool: """Disconnect from Zulip.""" self._running = False self._mark_disconnected() if self._poll_task: self._poll_task.cancel() try: await self._poll_task except asyncio.CancelledError: pass self._poll_task = None if self._client and self._queue_id: try: await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.deregister(self._queue_id) ) except Exception: pass self._client = None self._queue_id = None self._seen_messages.clear() self._pending_replies.clear() logger.info("[%s] Disconnected", self.name) return True # ── Polling loop ──────────────────────────────────────────────────── async def _poll_loop(self) -> None: """Poll the Zulip event queue for new messages, with reconnect.""" backoff_idx = 0 loop_start: float = 0.0 while self._running: try: loop_start = time.monotonic() await self._poll_once() if time.monotonic() - loop_start < 60.0: backoff_idx = 0 await asyncio.sleep(self._poll_interval) except asyncio.CancelledError: return except Exception as e: if not self._running: return logger.warning("[%s] Poll error: %s", self.name, e) if "BAD_EVENT_QUEUE_ID" in str(e): logger.info("[%s] Queue expired, re-registering...", self.name) if await self._reregister_queue(): backoff_idx = 0 continue delay = RECONNECT_BACKOFF[ min(backoff_idx, len(RECONNECT_BACKOFF) - 1) ] logger.info("[%s] Retrying in %ds...", self.name, delay) await asyncio.sleep(delay) backoff_idx += 1 async def _poll_once(self) -> None: """Retrieve and process one batch of events.""" if not self._client or not self._queue_id: return data = await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.get_events( queue_id=self._queue_id, last_event_id=self._last_event_id, ), ) if not data or not isinstance(data, dict): return if data.get("result") == "error": msg = data.get("msg", "") if "BAD_EVENT_QUEUE_ID" in msg or "queue_id" in msg.lower(): raise RuntimeError(f"BAD_EVENT_QUEUE_ID: {msg}") return events = data.get("events", []) if not events: return for event in events: event_id = event.get("id", 0) if event_id > self._last_event_id: self._last_event_id = event_id await self._on_event(event) async def _reregister_queue(self) -> bool: """Re-register the event queue after expiry.""" try: queue_result = await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.register( event_types=["message"], fetch_event_types=["message"], ), ) self._queue_id = queue_result.get("queue_id") self._last_event_id = queue_result.get("last_event_id", -1) if self._queue_id: logger.info( "[%s] Queue re-registered: %s", self.name, self._queue_id ) return True except Exception as e: logger.error("[%s] Queue re-registration failed: %s", self.name, e) return False # ── Event processing ──────────────────────────────────────────────── async def _on_event(self, event: Dict[str, Any]) -> None: """Process a single Zulip event.""" if event.get("type") != "message": return msg = event.get("message", {}) if not msg: return # Skip own messages (echo loop prevention) sender_email = msg.get("sender_email", "") if sender_email == self._email: return # Deduplicate event_id = str(event.get("id", "")) if self._is_duplicate(event_id): return # DM-first architecture (ADR-001): only process private messages if msg.get("type") == "private": await self._on_dm(msg) return async def _on_dm(self, msg: Dict[str, Any]) -> None: """Process an incoming DM and dispatch to the gateway.""" sender_id = str(msg.get("sender_id", "")) sender_email = str(msg.get("sender_email", "unknown")) sender_name = str(msg.get("sender_full_name", sender_email)) content = str(msg.get("content", "")).strip() if not content or not sender_id: return logger.info( "[%s] DM from %s (%s): %.60s", self.name, sender_name, sender_email, content, ) # Send a placeholder for streaming UX placeholder_msg_id = await self._send_placeholder(sender_id) # Register pending reply so send_message can edit the placeholder self._pending_replies[sender_id] = { "placeholder_msg_id": placeholder_msg_id, "type": "private", } # Fire-and-forget typing indicator asyncio.ensure_future( self._send_typing_indicator(int(sender_id), "start") ) # Build the session source source = self.build_source( chat_id=sender_id, chat_name=sender_email, chat_type="dm", user_id=sender_email, user_name=sender_name, message_id=str(msg.get("id", "")), ) # Build the MessageEvent for the gateway now = datetime.now(tz=timezone.utc) message_event = MessageEvent( text=content, message_type=MessageType.TEXT, source=source, message_id=str(msg.get("id", "")), raw_message=msg, timestamp=now, ) logger.debug("[%s] Dispatching DM from %s", self.name, sender_email) await self.handle_message(message_event) def _is_duplicate(self, msg_id: str) -> bool: """Dedup check with sliding window.""" now = time.time() stale = [ mid for mid, ts in self._seen_messages.items() if now - ts > DEDUP_WINDOW_SECONDS ] for mid in stale: del self._seen_messages[mid] if len(self._seen_messages) > DEDUP_MAX_SIZE: oldest = sorted( self._seen_messages.keys(), key=lambda k: self._seen_messages[k] )[:100] for k in oldest: del self._seen_messages[k] if msg_id in self._seen_messages: return True self._seen_messages[msg_id] = now return False # ── Message delivery ──────────────────────────────────────────────── async def send_message( self, chat_id: str, content: str, *, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, edit_message_id: Optional[str] = None, ) -> SendResult: """Send a DM reply. If a pending placeholder exists for this chat, edits it with the final response (streaming UX). Otherwise sends a fresh message. """ # Check for pending placeholder to finalize pending = self._pending_replies.pop(chat_id, None) if pending and pending.get("placeholder_msg_id"): return await self._edit_message( pending["placeholder_msg_id"], content[:MAX_MESSAGE_LENGTH], chat_id, ) return await self._send_fresh(chat_id, content) async def send_typing(self, chat_id: str, metadata=None) -> None: """Send typing indicator. No-op; managed via event flow.""" pass async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """Return basic info about a Zulip chat.""" return {"name": chat_id, "type": "dm"} # ── Internal helpers ──────────────────────────────────────────────── async def _send_placeholder(self, chat_id: str) -> Optional[str]: """Send a 'Thinking...' placeholder message. Returns message ID.""" if not self._client: return None placeholder = PLACEHOLDERS[ self._next_reply_id % len(PLACEHOLDERS) ] try: result = await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.send_message({ "type": "private", "to": chat_id, "content": placeholder, }), ) if result.get("result") == "success": msg_id = str(result.get("id", "")) logger.debug("[%s] Placeholder sent to %s (msg_id=%s)", self.name, chat_id, msg_id) return msg_id except Exception as e: logger.debug("[%s] Placeholder send failed: %s", self.name, e) return None async def _send_typing_indicator(self, user_id: int, operation: str) -> None: """Send typing start/stop via Zulip API.""" if not self._client: return try: await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.send_typing_notification( recipients=[user_id], operation=operation, ), ) except Exception: pass async def _edit_message( self, message_id: str, content: str, chat_id: str ) -> SendResult: """Edit an existing Zulip message (finalize a placeholder).""" if not self._client: return SendResult(success=False, error="Client not connected") try: await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.update_message({ "message_id": message_id, "content": content, }), ) return SendResult(success=True, message_id=message_id) except Exception as e: logger.warning( "[%s] Edit failed, sending fresh: %s", self.name, e ) return await self._send_fresh(chat_id, content) async def _send_fresh(self, chat_id: str, content: str) -> SendResult: """Send a fresh DM to Zulip.""" if not self._client: return SendResult(success=False, error="Client not connected") try: truncated = content[:MAX_MESSAGE_LENGTH] result = await asyncio.get_event_loop().run_in_executor( None, lambda: self._client.send_message({ "type": "private", "to": chat_id, "content": truncated, }), ) if result.get("result") == "success": return SendResult( success=True, message_id=str(result.get("id", "")), ) return SendResult( success=False, error=result.get("msg", "Send failed"), ) except Exception as e: logger.error("[%s] Send error: %s", self.name, e) return SendResult(success=False, error=str(e)) # ── Plugin registration ──────────────────────────────────────────────────── 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_EMAIL", "ZULIP_API_KEY", "ZULIP_SITE"], install_hint="pip install zulip httpx", env_enablement_fn=_env_enablement, cron_deliver_env_var="ZULIP_HOME_CHANNEL", standalone_sender_fn=_standalone_send, allowed_users_env="ZULIP_ALLOWED_USERS", allow_all_env="ZULIP_ALLOW_ALL_USERS", max_message_length=MAX_MESSAGE_LENGTH, emoji="💬", pii_safe=False, allow_update_command=True, platform_hint=( "You are communicating via Zulip direct messages. " "Your responses are delivered through Zulip's API. " "Use markdown formatting for rich responses. " f"Keep messages under {MAX_MESSAGE_LENGTH} characters (Zulip limit)." ), )