""" Zulip platform adapter (Hermes plugin). 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. 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 datetime import datetime, timezone from typing import Any, Dict, List, Optional try: import httpx HTTPX_AVAILABLE = True except ImportError: HTTPX_AVAILABLE = False httpx = None from gateway.config import 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] # 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. """ def __init__(self, config: PlatformConfig): platform_name = "zulip" super().__init__(config=config, platform=platform_name) 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._all_bots_user_id: int = int( extra.get("all_bots_user_id") or os.getenv("ZULIP_ALL_BOTS_USER_ID", str(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._http_client: Optional[httpx.AsyncClient] = None self._seen_message_ids: Dict[str, float] = {} self._pending_replies: Dict[str, int] = {} # msg_id -> placeholder zulip_msg_id # Derived identity for echo-loop prevention self._bot_user_id: Optional[int] = None self._bot_email: str = self._email # ------------------------------------------------------------------ # 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) # 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") self._mark_connected() logger.info( "[%s] Connected to %s as %s (queue=%s, bot_id=%s)", self.name, self._site, self._email, self._queue_id, self._bot_user_id, ) # Start poll loop self._poll_task = asyncio.create_task(self._poll_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 the poll loop.""" 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() for event in events: await self._process_zulip_event(event) backoff_idx = 0 except asyncio.CancelledError: break except Exception as e: if not self._running: break 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) logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id) except Exception as e: logger.error("[%s] Reconnect failed: %s", self.name, e) # ------------------------------------------------------------------ # 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 # Deduplication msg_id = f"{event.get('id', '')}:{msg.get('id', '')}" if self._is_duplicate(msg_id): return # 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: 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): 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, ) 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"): return SendResult( success=True, platform="zulip", chat_id=chat_id, message_id=str(result["id"]), ) 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 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*") # ------------------------------------------------------------------ # 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 and time window.""" now = time.time() window = 300 # 5 minutes max_size = 1000 # Clean old entries if len(self._seen_message_ids) > max_size: cutoff = now - window self._seen_message_ids = { k: v for k, v in self._seen_message_ids.items() if v > cutoff } if msg_id in self._seen_message_ids: return True self._seen_message_ids[msg_id] = now return False # ------------------------------------------------------------------ # 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." ), )