diff --git a/hermes-zulip-plugin/__init__.py b/hermes-zulip-plugin/__init__.py new file mode 100644 index 0000000..d4f1d7b --- /dev/null +++ b/hermes-zulip-plugin/__init__.py @@ -0,0 +1,3 @@ +from .adapter import register + +__all__ = ["register"] diff --git a/hermes-zulip-plugin/adapter.py b/hermes-zulip-plugin/adapter.py new file mode 100644 index 0000000..3dfaa30 --- /dev/null +++ b/hermes-zulip-plugin/adapter.py @@ -0,0 +1,616 @@ +"""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)." + ), + ) diff --git a/hermes-zulip-plugin/plugin.yaml b/hermes-zulip-plugin/plugin.yaml new file mode 100644 index 0000000..6664474 --- /dev/null +++ b/hermes-zulip-plugin/plugin.yaml @@ -0,0 +1,47 @@ +name: zulip-platform +label: Zulip +kind: platform +version: 1.0.0 +description: > + Zulip messaging gateway adapter for Hermes Agent. + Connects to any Zulip server, registers an event queue, and polls for + incoming messages. Handles DMs first (ADR-001), with @mention support + planned. Uses the Zulip Python SDK for API access. + + DM-first architecture: private messages route directly into the Hermes + agent's session, so the same personality and conversation continuity + is maintained between TUI, CLI, and Zulip. + + Features: event queue polling, typing indicators, placeholder→edit + streaming, exponential backoff reconnection, health endpoint. +author: Syslog Solution LLC +requires_env: + - name: ZULIP_EMAIL + description: "Zulip bot email address (e.g. tanko-bot@chat.sysloggh.net)" + prompt: "Zulip bot email" + password: false + - name: ZULIP_API_KEY + description: "Zulip bot API key" + prompt: "Zulip API key" + password: true + - name: ZULIP_SITE + description: "Zulip server URL (e.g. https://chat.sysloggh.net)" + prompt: "Zulip server URL" + password: false +optional_env: + - name: ZULIP_ALLOWED_USERS + description: "Comma-separated Zulip user IDs or emails allowed to DM the agent" + prompt: "Allowed users (comma-separated emails or IDs)" + password: false + - name: ZULIP_ALLOW_ALL_USERS + description: "Allow any user to DM the bot (disables allowlist)" + prompt: "Allow all users? (true/false)" + password: false + - name: ZULIP_HOME_CHANNEL + description: "Default recipient for cron / notification delivery" + prompt: "Home channel user email (or empty)" + password: false + - name: ZULIP_HOME_CHANNEL_NAME + description: "Human label for the home channel" + prompt: "Home channel display name" + password: false diff --git a/hermes-zulip-plugin/requirements.txt b/hermes-zulip-plugin/requirements.txt index e6a2a4f..bd399be 100644 --- a/hermes-zulip-plugin/requirements.txt +++ b/hermes-zulip-plugin/requirements.txt @@ -1,3 +1,2 @@ zulip>=0.9.0 -zulip-bots>=0.9.0 -pyyaml>=6.0 +httpx>=0.27.0 diff --git a/hermes-zulip-plugin/src/hermes_zulip/__init__.py b/hermes-zulip-plugin/src/hermes_zulip/__init__.py deleted file mode 100644 index a593145..0000000 --- a/hermes-zulip-plugin/src/hermes_zulip/__init__.py +++ /dev/null @@ -1,11 +0,0 @@ -""" -Hermes Zulip Plugin — Core adapter for Hermes Python agents - -Implements BasePlatformAdapter to connect Hermes agents (Tanko, Mumuni, -Koonimo, Koby) to the Sysloggh Zulip agent mesh. - -Path: hermes-zulip-plugin/src/hermes_zulip/ -Config: config.yaml (per-agent, deployed alongside plugin) -""" - -__version__ = "0.1.0" diff --git a/hermes-zulip-plugin/src/hermes_zulip/adapter.py b/hermes-zulip-plugin/src/hermes_zulip/adapter.py deleted file mode 100644 index fea2470..0000000 --- a/hermes-zulip-plugin/src/hermes_zulip/adapter.py +++ /dev/null @@ -1,133 +0,0 @@ -# Core adapter implements BasePlatformAdapter for Zulip -# See ADR-005 for @mention detection, ADR-009 for error handling -# Full implementation in issue #5. - -import logging -import time -import re -import threading -from typing import Any, Dict, Optional - -logger = logging.getLogger(__name__) - -class ZulipAdapter: - """Zulip adapter for Hermes agents. Connects to Zulip, listens for - @mentions in #agent-hub, routes messages via BasePlatformAdapter.""" - - # Regex to strip Zulip mention artifacts from message bodies (ADR-008) - MENTION_CLEANER = re.compile(r'@\*\*[^*]+\*\*') - - def __init__(self, config: Dict[str, Any]) -> None: - self.config = config - self.connected = False - self._client = None - self._thread = None - self._stop_event = threading.Event() - - async def connect(self) -> None: - """Establish connection to Zulip and start the event loop.""" - if self.connected: - logger.info("Already connected to Zulip.") - return - - try: - from zulip import Client - server_url = self.config['zulip']['server_url'] - email = self.config['zulip']['email'] - api_key = self.config['zulip']['api_key'] - - self._client = Client(server_url=server_url, email=email, api_key=api_key) - self.connected = True - logger.info(f"Connected to Zulip: {server_url} as {email}") - - # Start the event loop in a background thread - self._stop_event.clear() - self._thread = threading.Thread(target=self._event_loop, daemon=True) - self._thread.start() - logger.info("Zulip event loop started.") - - except Exception as e: - logger.error(f"Failed to connect to Zulip: {e}") - self.connected = False - raise - - async def disconnect(self) -> None: - """Disconnect from Zulip and stop the event loop.""" - if not self.connected: - return - self._stop_event.set() - if self._thread: - self._thread.join(timeout=5) - self.connected = False - logger.info("Disconnected from Zulip.") - - async def send_message(self, topic: str, content: str) -> Dict[str, Any]: - """Send a message to Zulip with retry logic (ADR-009).""" - max_retries = self.config.get('error_handling', {}).get('retry_count', 3) - retry_delay = self.config.get('error_handling', {}).get('retry_delay_seconds', 5) - - for attempt in range(max_retries): - try: - result = self._client.send_message({ - 'type': 'stream', - 'to': self.config['zulip']['stream'], - 'subject': topic, - 'content': content, - }) - logger.info(f"Message sent to {self.config['zulip']['stream']} > {topic} (attempt {attempt + 1})") - return result - except Exception as e: - logger.warning(f"Failed to send message (attempt {attempt + 1}): {e}") - if attempt < max_retries - 1: - time.sleep(retry_delay) - else: - raise - - async def on_event(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """Process incoming Zulip events and route via BasePlatformAdapter.""" - if event.get('type') != 'message': - return None - - msg = event.get('message', {}) - mentioned_users = msg.get('mentioned_users', False) - - # Only process messages that mention this bot (ADR-005) - if not mentioned_users: - return None - - # Strip Zulip formatting artifacts - body = msg.get('content', '') - clean_body = self.MENTION_CLEANER.sub('', body) - clean_body = clean_body.strip() - - # Construct the MessageEvent for BasePlatformAdapter - return { - 'type': 'MessageEvent', - 'sender': msg.get('sender_full_name', 'Unknown'), - 'body': clean_body, - 'topic': msg.get('subject', 'Unknown Topic'), - 'stream': msg.get('stream', 'Unknown Stream'), - 'timestamp': msg.get('timestamp', 0), - } - - def _event_loop(self) -> None: - """Background thread to listen for Zulip events.""" - try: - self._client.call_on_each_message( - lambda event: self._process_event(event), - ) - except Exception as e: - logger.error(f"Event loop crashed: {e}") - self.connected = False - - def _process_event(self, event: Dict[str, Any]) -> None: - """Bridge the synchronous event to the async on_event handler.""" - import asyncio - try: - loop = asyncio.get_event_loop() - if loop.is_running(): - asyncio.create_task(self.on_event(event)) - else: - asyncio.run(self.on_event(event)) - except Exception as e: - logger.error(f"Error processing event: {e}")