""" Zulip platform adapter (Hermes plugin) — Gen 4. 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 4 Improvements (2026-06-27): 1. HTTP client recreation on reconnect — new httpx.AsyncClient() after queue expiry or sustained failures to prevent CLOSE-WAIT socket leaks 2. Heartbeat logging — periodic "still alive" message to prove poll loop is running 3. Log sanitization — truncates 502/error HTML to first 80 chars (no Netbird HTML spam) 4. Connection recovery — creates fresh HTTP client after N empty poll cycles 5. Poll silence detection — logs warnings when no events received for extended period 6. Connection reuse limit — periodic client recreation to prevent connection pool exhaustion 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 # Gen 4: Connection pool health HEARTBEAT_INTERVAL = 300 # Log heartbeat every 5 minutes MAX_CONSECUTIVE_EMPTY_POLLS = 500 # Reset client after this many empty polls (~25min at 3s interval). Raised from 20 to prevent idle bots from cycling reconnections. CLIENT_REUSE_LIMIT = 500 # Create fresh client every N polls to prevent connection leaks POLL_SILENCE_WARN_INTERVAL = 60 # Warn if no events received for this many seconds MAX_ERROR_LOG_LEN = 80 # Truncate error response bodies to avoid HTML spam # 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)) ) # --- Gen 4: Connection health --- self._consecutive_empty_polls: int = 0 self._total_poll_count: int = 0 self._last_event_received_time: float = time.time() self._last_heartbeat_time: float = 0.0 self._client_pool_reset_count: int = 0 # --- State --- # Support both ZULIP_SITE and ZULIP_URL env vars if not self._site: self._site = (os.getenv("ZULIP_URL", "") or "").rstrip("/") 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 = 300 # 5 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, **kwargs: Any) -> bool: """Connect to Zulip and register an event queue. Accepts **kwargs for Hermes Gateway compatibility (e.g., is_reconnect). """ 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: await self._create_http_client() self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat() # Reset connection health counters self._consecutive_empty_polls = 0 self._total_poll_count = 0 self._last_event_received_time = time.time() # 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") # Resolve bot user_id if register endpoint didn't provide it # (common on some Zulip versions) if self._bot_user_id is None: await self._resolve_bot_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._total_poll_count += 1 self._health_stats["last_poll_at"] = datetime.now( timezone.utc ).isoformat() if events: now = time.time() self._consecutive_empty_polls = 0 self._last_event_received_time = now 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 ) else: # Gen 4: Track consecutive empty polls to detect silent failures self._consecutive_empty_polls += 1 # Gen 4: Detect prolonged silence (no events for too long) silence_duration = time.time() - self._last_event_received_time if silence_duration > POLL_SILENCE_WARN_INTERVAL: logger.warning( "[%s] No events received for %.0fs " "(consecutive_empty=%d, total_polls=%d, " "queue=%s)", self.name, silence_duration, self._consecutive_empty_polls, self._total_poll_count, self._queue_id, ) # Gen 4: Reset HTTP client if too many empty polls in a row # (connection pool likely stuck in CLOSE-WAIT) if ( self._consecutive_empty_polls >= MAX_CONSECUTIVE_EMPTY_POLLS ): logger.warning( "[%s] %d consecutive empty polls — " "recreating HTTP client and reconnecting", self.name, self._consecutive_empty_polls, ) await self._reconnect_with_fresh_client() backoff_idx = 0 self._consecutive_empty_polls = 0 continue # Gen 4: Periodic client refresh to prevent connection leaks if self._total_poll_count % CLIENT_REUSE_LIMIT == 0: logger.info( "[%s] Client reuse limit reached (%d polls) — " "creating fresh HTTP client", self.name, self._total_poll_count, ) old_client = self._http_client await self._create_http_client() if old_client: try: await old_client.aclose() except Exception: pass # Clear stale errors on successful poll cycle self._health_stats["last_error_msg"] = None self._health_stats["last_error_at"] = None # Gen 4: Heartbeat logging — prove poll loop is alive now = time.time() if now - self._last_heartbeat_time > HEARTBEAT_INTERVAL: self._last_heartbeat_time = now silence = now - self._last_event_received_time logger.info( "[%s] Heartbeat — polls=%d empty=%d " "silence=%.0fs errors=%d reconnects=%d " "queue=%s", self.name, self._total_poll_count, self._consecutive_empty_polls, silence, self._health_stats["poll_errors"], self._health_stats["reconnects"], self._queue_id, ) 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_with_fresh_client() 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. Raises RuntimeError with BAD_EVENT_QUEUE_ID when the queue expires, so _poll_forever can reconnect. """ if not self._queue_id: return [] resp, raw_status = 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", }, ) # Detect queue expiry from HTTP response if raw_status == 400: raise RuntimeError("BAD_EVENT_QUEUE_ID: queue expired") 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 _create_http_client(self) -> None: """Create a fresh httpx client, closing the old one if it exists. Gen 4: This ensures a clean connection pool after sustained failures or queue expiry, preventing CLOSE-WAIT socket leaks. """ old_client = self._http_client self._http_client = httpx.AsyncClient( timeout=httpx.Timeout(30.0, connect=15.0), limits=httpx.Limits( max_keepalive_connections=5, max_connections=10, keepalive_expiry=60.0, ), ) self._client_pool_reset_count += 1 if old_client: try: await old_client.aclose() except Exception: pass 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) self._consecutive_empty_polls = 0 self._last_event_received_time = time.time() except Exception as e: logger.error("[%s] Reconnect failed: %s", self.name, e) async def _reconnect_with_fresh_client(self) -> None: """Re-register the event queue with a fresh HTTP client. Gen 4: Creates a new httpx client to break out of stuck connection pools (CLOSE-WAIT sockets from previous failures). """ logger.info("[%s] Reconnecting with fresh HTTP client", self.name) await self._create_http_client() await self._reconnect() # ------------------------------------------------------------------ # 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 5: Bot user ID resolution # ------------------------------------------------------------------ async def _resolve_bot_user_id(self) -> None: """Fetch the bot's own user ID from /api/v1/users/me. The /register endpoint does not always return user_id on all Zulip server versions. This ensures @mention detection works by resolving it from the identity endpoint. """ try: resp, _ = await self._api_call("GET", "/api/v1/users/me") if resp: uid = resp.get("user_id") if uid is not None: self._bot_user_id = int(uid) logger.info( "[%s] Resolved bot user_id=%s from /users/me", self.name, uid, ) except Exception as e: logger.debug( "[%s] Could not resolve bot user_id: %s", self.name, e ) # ------------------------------------------------------------------ # 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, **kwargs: Any, ) -> 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. Accepts **kwargs for Hermes Gateway compatibility. """ # 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, message_id=msg_id, ) # Send actual content result = await self._send_api_call(payload) if result and result.get("id"): msg_id = str(result["id"]) self._health_stats["send_count"] += 1 return SendResult( success=True, message_id=msg_id, ) self._health_stats["send_errors"] += 1 return SendResult( success=False, error="Failed to send message", ) async def edit_message( self, chat_id: str, message_id: str, content: str, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> SendResult: """Edit a previously sent Zulip message (for placeholder replacement). Returns SendResult for Hermes Gateway streaming interface compatibility. Accepts **kwargs (e.g., finalize=True) — extra kwargs are silently ignored. Note: Zulip uses POST /api/v1/messages/{message_id} for editing. """ truncated = _truncate(content) payload = { "content": truncated, } try: # Zulip API: PATCH /api/v1/messages/{message_id} with content in body resp, status = await self._api_call( "PATCH", f"/api/v1/messages/{int(message_id)}", data=payload, ) if resp: self._health_stats["send_count"] += 1 return SendResult( success=True, message_id=str(resp.get("id", message_id)), ) self._health_stats["send_errors"] += 1 return SendResult( success=False, error=f"Edit failed (status={status})", ) except Exception as e: logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e) self._health_stats["send_errors"] += 1 return SendResult( success=False, error=str(e), ) 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"}, ) # typing indicator failure is non-critical 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"}, ) # typing indicator failure is non-critical 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, **kwargs: Any, ) -> SendResult: """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*", **kwargs) # ------------------------------------------------------------------ # 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() stats["total_polls"] = self._total_poll_count stats["consecutive_empty_polls"] = self._consecutive_empty_polls stats["client_pool_resets"] = self._client_pool_reset_count stats["silence_seconds"] = round(time.time() - self._last_event_received_time) # 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, ) -> Tuple[Optional[Dict], int]: """Make an API call to Zulip. Returns (response_json, status_code). status_code is useful for callers that need to detect specific HTTP errors like 400 (BAD_EVENT_QUEUE_ID). Returns (None, 0) on transport errors. """ if not self._http_client: return None, 0 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, 0 if response.status_code >= 400: # Gen 4: Truncate error bodies to prevent HTML spam body = response.text[:MAX_ERROR_LOG_LEN].replace("\n", " ") logger.warning( "[%s] API %s %s: %d %s", self.name, method, path, response.status_code, body, ) return None, response.status_code return response.json(), response.status_code except httpx.TimeoutException: logger.debug("[%s] Timeout on %s %s", self.name, method, path) return None, 0 except Exception as e: logger.warning("[%s] API error %s %s: %s", self.name, method, path, e) return None, 0 async def _send_api_call(self, payload: Dict) -> Optional[Dict]: """Send a message to Zulip.""" result, _ = await self._api_call("POST", "/api/v1/messages", data=payload) return result # ------------------------------------------------------------------ # 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() or os.getenv("ZULIP_URL", "").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", "") or os.getenv("ZULIP_URL", "") 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() or os.getenv("ZULIP_URL", "").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." ), )