diff --git a/hermes-zulip-plugin/plugin.yaml b/hermes-zulip-plugin/plugin.yaml new file mode 100644 index 0000000..ea65517 --- /dev/null +++ b/hermes-zulip-plugin/plugin.yaml @@ -0,0 +1,61 @@ +name: zulip-platform +label: Zulip +kind: platform +manifest_version: 1 +version: 1.0.0 +description: > + Zulip gateway adapter for Hermes Agent. Connects to a self-hosted or cloud + Zulip realm via the REST API — an event queue (register + long-poll) for + receiving and POST /messages for sending — relaying messages between Zulip + channels (streams)/topics and direct messages and the Hermes agent. + Supports topic-aware replies, native file uploads, channel-scoped + allowlists, @mention gating via message flags, and home-channel cron + delivery. No Zulip SDK required — uses aiohttp. +author: wachtelhund +requires_env: + - name: ZULIP_SITE + description: "Zulip realm URL (e.g. https://example.zulipchat.com)" + prompt: "Zulip site URL" + password: false + url: "https://zulip.com/api/api-keys" + - name: ZULIP_EMAIL + description: "Bot email address" + prompt: "Bot email" + password: false + - name: ZULIP_API_KEY + description: "Bot API key" + prompt: "Zulip bot API key" + password: true +optional_env: + - name: ZULIP_ALLOWED_USERS + description: "Comma-separated sender emails/ids allowed to talk to the bot" + prompt: "Allowed users (comma-separated)" + password: false + - name: ZULIP_ALLOW_ALL_USERS + description: "Allow any Zulip user to trigger the bot (dev only)" + prompt: "Allow all users? (true/false)" + password: false + - name: ZULIP_HOME_CHANNEL + description: "Default target for cron / notification delivery (e.g. general/hermes)" + prompt: "Home channel (stream/topic)" + password: false + - name: ZULIP_REQUIRE_MENTION + description: "Require @bot mention in channels (default true). Set false for free-response everywhere." + prompt: "Require @mention? (true/false)" + password: false + - name: ZULIP_FREE_RESPONSE_CHANNELS + description: "Comma-separated stream ids/names where @mention is not required." + prompt: "Free-response streams (comma-separated)" + password: false + - name: ZULIP_ALLOWED_CHANNELS + description: "If set, the bot only responds in these stream ids/names (whitelist)." + prompt: "Allowed streams (comma-separated)" + password: false + - name: ZULIP_ALL_PUBLIC_STREAMS + description: "Subscribe the event queue to all public streams (default false)." + prompt: "Listen to all public streams? (true/false)" + password: false + - name: ZULIP_DEFAULT_TOPIC + description: "Topic used when a target has none (default 'general')." + prompt: "Default topic" + password: false diff --git a/hermes-zulip-plugin/src/__init__.py b/hermes-zulip-plugin/src/__init__.py new file mode 100644 index 0000000..56625c5 --- /dev/null +++ b/hermes-zulip-plugin/src/__init__.py @@ -0,0 +1,13 @@ +"""Zulip gateway plugin for Hermes Agent.""" +try: + # Normal path: Hermes loads this as a package (``hermes_plugins.…``). + from .adapter import register +except ImportError: # pragma: no cover + # Imported as a top-level module (e.g. by a test collector that treats + # the repo root as a package). Fall back to an absolute import. + import os + import sys + sys.path.insert(0, os.path.dirname(__file__)) + from adapter import register + +__all__ = ["register"] diff --git a/hermes-zulip-plugin/src/adapter.py b/hermes-zulip-plugin/src/adapter.py new file mode 100644 index 0000000..7542b9c --- /dev/null +++ b/hermes-zulip-plugin/src/adapter.py @@ -0,0 +1,976 @@ +"""Zulip gateway adapter for Hermes Agent. + +Connects to a Zulip realm (self-hosted or zulipchat.com) and relays messages +between Zulip channels (streams) / direct messages and the Hermes agent. + +Both transports use Zulip's REST API over ``aiohttp`` (already a Hermes +dependency — no ``zulip`` SDK required): + +* **Receiving** — register an event queue (``POST /api/v1/register``) and + long-poll it (``GET /api/v1/events``). The queue delivers ``message`` + events; the bot's ``mentioned`` flag drives channel gating. A + ``BAD_EVENT_QUEUE_ID`` (expired queue) transparently re-registers. +* **Sending** — ``POST /api/v1/messages`` (stream + topic, or direct) and + ``POST /api/v1/user_uploads`` for file attachments. REST is also used + out-of-process by cron delivery via :func:`_standalone_send`. + +Authentication is HTTP Basic with the bot's email + API key. + +Chat ids are encoded so replies route back to the right place: + + s:: a stream message under a topic (chat_type=channel) + d:,,… a direct message (chat_type=dm) + +Environment variables (set in ``~/.hermes/.env``): + + ZULIP_SITE Realm URL (e.g. https://example.zulipchat.com) + ZULIP_EMAIL Bot email + ZULIP_API_KEY Bot API key + +Optional: + + ZULIP_ALLOWED_USERS Comma-separated sender emails/ids allowed to talk to the bot + ZULIP_ALLOW_ALL_USERS Allow any user (dev only) + ZULIP_HOME_CHANNEL Default target for cron delivery (e.g. "general/hermes" or "s:5:hermes") + ZULIP_REQUIRE_MENTION Require @bot mention in channels (default true) + ZULIP_FREE_RESPONSE_CHANNELS Stream ids/names where @mention is not required + ZULIP_ALLOWED_CHANNELS If set, the bot only responds in these stream ids/names + ZULIP_ALL_PUBLIC_STREAMS Subscribe the queue to all public streams (default false) + ZULIP_DEFAULT_TOPIC Topic used when a target has none (default "general") +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import re +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.helpers import MessageDeduplicator +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + SendResult, +) + +logger = logging.getLogger(__name__) + +# Identifier used in config.yaml (gateway.platforms.zulip) and by +# ``Platform("zulip")``. Registered at import-time via register(). +PLATFORM_NAME = "zulip" + +# Zulip's hard message limit is 10000 chars; 9000 leaves headroom for the +# chunk indicator and any markdown the agent adds. +MAX_MESSAGE_LENGTH = 9000 + +# Long-poll client timeout. Zulip holds GET /events open and emits a +# heartbeat well within this window, so a timeout just means "re-poll". +_LONGPOLL_TIMEOUT = 90.0 + +_RECONNECT_BASE_DELAY = 2.0 +_RECONNECT_MAX_DELAY = 60.0 +_RECONNECT_JITTER = 0.2 + + +def check_zulip_requirements() -> bool: + """Return True if the Zulip adapter can be used.""" + site = os.getenv("ZULIP_SITE", "") or os.getenv("ZULIP_URL", "") + email = os.getenv("ZULIP_EMAIL", "") + api_key = os.getenv("ZULIP_API_KEY", "") + if not site: + logger.debug("Zulip: ZULIP_SITE not set") + return False + if not email or not api_key: + logger.debug("Zulip: ZULIP_EMAIL and ZULIP_API_KEY required") + return False + try: + import aiohttp # noqa: F401 + return True + except ImportError: + logger.warning("Zulip: aiohttp not installed") + return False + + +def _parse_target(chat_id: str, default_topic: str = "general") -> Tuple[str, Any, Optional[str]]: + """Decode a chat id / home-channel string into a send target. + + Returns ``(kind, to, topic)`` where *kind* is ``"stream"`` or + ``"direct"``. Accepted forms: + + d:1,2,3 -> ("direct", [1, 2, 3], None) + s:: -> ("stream", , ) + / -> ("stream", , ) (friendly) + -> ("stream", , default_topic) + """ + cid = (chat_id or "").strip() + if cid.startswith("d:"): + ids: List[Any] = [] + for part in cid[2:].split(","): + part = part.strip() + if not part: + continue + ids.append(int(part) if part.isdigit() else part) + return "direct", ids, None + if cid.startswith("s:"): + rest = cid[2:] + stream, _, topic = rest.partition(":") + stream = stream.strip() + to: Any = int(stream) if stream.isdigit() else stream + return "stream", to, (topic or default_topic) + if "/" in cid: + stream, topic = cid.split("/", 1) + return "stream", stream.strip(), (topic.strip() or default_topic) + return "stream", cid, default_topic + + +class ZulipAdapter(BasePlatformAdapter): + """Gateway adapter for Zulip (self-hosted or zulipchat.com).""" + + def __init__(self, config: PlatformConfig): + super().__init__(config, Platform(PLATFORM_NAME)) + + extra = config.extra or {} + self._site: str = ( + extra.get("site") + or os.getenv("ZULIP_SITE", "") + or os.getenv("ZULIP_URL", "") + ).rstrip("/") + self._email: str = extra.get("email") or os.getenv("ZULIP_EMAIL", "") + self._api_key: str = config.token or os.getenv("ZULIP_API_KEY", "") + + self._bot_user_id: Optional[int] = None + self._bot_email: str = "" + self._bot_full_name: str = "" + + self._session: Any = None + self._queue_id: Optional[str] = None + self._last_event_id: int = -1 + self._poll_task: Optional[asyncio.Task] = None + self._closing = False + + self._default_topic: str = ( + extra.get("default_topic") or os.getenv("ZULIP_DEFAULT_TOPIC", "general") + ) + + self._dedup = MessageDeduplicator() + + # ------------------------------------------------------------------ + # HTTP helpers + # ------------------------------------------------------------------ + + def _auth(self): + import aiohttp + return aiohttp.BasicAuth(self._email, self._api_key) + + async def _api_get(self, path: str, params: Optional[Dict[str, Any]] = None, timeout: float = 30.0) -> Dict[str, Any]: + import aiohttp + url = f"{self._site}/api/v1/{path.lstrip('/')}" + try: + async with self._session.get( + url, params=params, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=timeout), + ) as resp: + data = await resp.json() + if resp.status >= 400: + logger.debug("Zulip GET %s -> %s: %s", path, resp.status, str(data)[:200]) + data.setdefault("_http_status", resp.status) + return data + # aiohttp raises a bare asyncio.TimeoutError (NOT a ClientError) on + # timeout — catch both so callers always get the error dict, never an + # unhandled exception. + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.error("Zulip GET %s network error: %s", path, exc) + return {"result": "error", "msg": str(exc)} + + async def _api_post(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]: + import aiohttp + url = f"{self._site}/api/v1/{path.lstrip('/')}" + try: + async with self._session.post( + url, data=payload, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + data = await resp.json() + if resp.status >= 400: + logger.error("Zulip POST %s -> %s: %s", path, resp.status, str(data)[:200]) + return data + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.error("Zulip POST %s network error: %s", path, exc) + return {"result": "error", "msg": str(exc)} + + async def _api_patch(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]: + import aiohttp + url = f"{self._site}/api/v1/{path.lstrip('/')}" + try: + async with self._session.patch( + url, data=payload, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=15), + ) as resp: + data = await resp.json() + if resp.status >= 400: + logger.debug("Zulip PATCH %s -> %s: %s", path, resp.status, str(data)[:200]) + return data + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.debug("Zulip PATCH %s network error: %s", path, exc) + return {"result": "error", "msg": str(exc)} + + async def _api_delete(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + import aiohttp + url = f"{self._site}/api/v1/{path.lstrip('/')}" + try: + async with self._session.delete( + url, params=params, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=15), + ) as resp: + return await resp.json() + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.debug("Zulip DELETE %s network error: %s", path, exc) + return {"result": "error", "msg": str(exc)} + + async def _upload(self, file_data: bytes, filename: str, content_type: str) -> Optional[str]: + """Upload a file, returning its absolute URL (or None).""" + import aiohttp + url = f"{self._site}/api/v1/user_uploads" + form = aiohttp.FormData() + form.add_field("file", file_data, filename=filename, content_type=content_type) + try: + async with self._session.post( + url, data=form, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=60), + ) as resp: + data = await resp.json() + if resp.status >= 400: + logger.error("Zulip upload -> %s: %s", resp.status, str(data)[:200]) + return None + rel = data.get("url") or data.get("uri") + if not rel: + return None + return rel if rel.startswith("http") else f"{self._site}{rel}" + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.error("Zulip upload network error: %s", exc) + return None + + # ------------------------------------------------------------------ + # Required overrides + # ------------------------------------------------------------------ + + async def connect(self, is_reconnect: bool = False) -> bool: + import aiohttp + + if not self._site or not self._email or not self._api_key: + logger.error("Zulip: ZULIP_SITE, ZULIP_EMAIL and ZULIP_API_KEY must be set") + return False + + self._session = aiohttp.ClientSession() + self._closing = False + + me = await self._api_get("users/me") + if me.get("result") != "success" or "user_id" not in me: + logger.error( + "Zulip: failed to authenticate — check ZULIP_SITE, ZULIP_EMAIL " + "and ZULIP_API_KEY (%s)", str(me.get("msg", ""))[:160], + ) + await self._session.close() + return False + + self._bot_user_id = me["user_id"] + self._bot_email = me.get("email", self._email) + self._bot_full_name = me.get("full_name", "") + logger.info( + "Zulip: authenticated as %s (#%s) on %s", + self._bot_full_name or self._bot_email, self._bot_user_id, self._site, + ) + + self._poll_task = asyncio.create_task(self._poll_loop()) + self._mark_connected() + return True + + async def disconnect(self) -> None: + self._closing = True + if self._poll_task and not self._poll_task.done(): + self._poll_task.cancel() + try: + await self._poll_task + except (asyncio.CancelledError, Exception): + pass + # Best-effort: release the server-side event queue. Zulip deletes a + # queue via DELETE /api/v1/events?queue_id=… (there is no POST form). + if self._queue_id and self._session and not self._session.closed: + try: + await self._api_delete("events", {"queue_id": self._queue_id}) + except Exception: + pass + if self._session and not self._session.closed: + await self._session.close() + logger.info("Zulip: disconnected") + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + if not content: + return SendResult(success=True) + + kind, to, topic = _parse_target(chat_id, self._default_topic) + formatted = self.format_message(content) + chunks = self.truncate_message(formatted, MAX_MESSAGE_LENGTH) + + last_id = None + for chunk in chunks: + payload: Dict[str, Any] = {"content": chunk} + if kind == "direct": + # "private" is the backward-compatible alias accepted by every + # Zulip version (the newer "direct" alias only landed in 7.0). + payload["type"] = "private" + payload["to"] = json.dumps(to) + else: + payload["type"] = "stream" + payload["to"] = str(to) + payload["topic"] = topic or self._default_topic + data = await self._api_post("messages", payload) + if data.get("result") != "success": + return SendResult(success=False, error=str(data.get("msg", "send failed"))) + last_id = data.get("id") + + return SendResult(success=True, message_id=str(last_id) if last_id else None) + + async def edit_message( + self, + chat_id: str, + message_id: str, + content: str, + **kwargs, + ) -> SendResult: + """Edit a previously-sent message. Used by the gateway stream consumer + for progressive message updates during agent streaming.""" + formatted = self.format_message(content) + data = await self._api_patch(f"messages/{message_id}", {"content": formatted}) + if data.get("result") != "success": + msg = str(data.get("msg", "edit failed")) + logger.debug("Zulip edit_message(%s) -> %s", message_id, msg) + return SendResult(success=False, error=msg) + return SendResult(success=True, message_id=message_id) + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + kind, to, topic = _parse_target(chat_id, self._default_topic) + if kind == "direct": + return {"name": "Direct message", "type": "dm", "chat_id": chat_id} + name = f"{to} > {topic}" if topic else str(to) + return {"name": name, "type": "channel", "chat_id": chat_id} + + # ------------------------------------------------------------------ + # Optional overrides — files are uploaded then linked into a message + # ------------------------------------------------------------------ + + async def send_image( + self, + chat_id: str, + image_url: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + return await self._send_url_as_file(chat_id, image_url, caption, "image") + + async def send_image_file( + self, + chat_id: str, + image_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + return await self._send_local_file(chat_id, image_path, caption) + + async def send_document( + self, + chat_id: str, + file_path: str, + caption: Optional[str] = None, + file_name: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + return await self._send_local_file(chat_id, file_path, caption, file_name) + + async def send_voice( + self, + chat_id: str, + audio_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + return await self._send_local_file(chat_id, audio_path, caption) + + async def send_video( + self, + chat_id: str, + video_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + return await self._send_local_file(chat_id, video_path, caption) + + def format_message(self, content: str) -> str: + """Zulip renders CommonMark. + + Convert inline image markdown into bare links — Zulip auto-previews + image URLs, and files are uploaded separately. + """ + return re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"\2", content) + + # ------------------------------------------------------------------ + # File helpers + # ------------------------------------------------------------------ + + async def _send_url_as_file( + self, chat_id: str, url: str, caption: Optional[str], kind: str = "file", + ) -> SendResult: + from tools.url_safety import is_safe_url + if not is_safe_url(url): + logger.warning("Zulip: blocked unsafe URL (SSRF protection)") + return await self.send(chat_id, f"{caption or ''}\n{url}".strip()) + + import aiohttp + fname = url.rsplit("/", 1)[-1].split("?")[0] or f"{kind}.png" + try: + async with self._session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + if resp.status >= 400: + return await self.send(chat_id, f"{caption or ''}\n{url}".strip()) + file_data = await resp.read() + ct = resp.content_type or "application/octet-stream" + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.warning("Zulip: failed to download %s: %s", url[:80], exc) + return await self.send(chat_id, f"{caption or ''}\n{url}".strip()) + + return await self._upload_and_send(chat_id, file_data, fname, ct, caption, fallback=url) + + async def _send_local_file( + self, chat_id: str, file_path: str, caption: Optional[str], file_name: Optional[str] = None, + ) -> SendResult: + import mimetypes + p = Path(file_path) + if not p.exists(): + logger.warning("Zulip: local file not found, skipping: %s", file_path) + return SendResult(success=True, message_id=None) + fname = file_name or p.name + ct = mimetypes.guess_type(fname)[0] or "application/octet-stream" + return await self._upload_and_send(chat_id, p.read_bytes(), fname, ct, caption) + + async def _upload_and_send( + self, chat_id: str, data: bytes, fname: str, ct: str, + caption: Optional[str], fallback: Optional[str] = None, + ) -> SendResult: + url = await self._upload(data, fname, ct) + if not url: + if fallback: + return await self.send(chat_id, f"{caption or ''}\n{fallback}".strip()) + return SendResult(success=False, error="File upload failed") + body = f"{caption + chr(10) if caption else ''}[{fname}]({url})" + return await self.send(chat_id, body) + + # ------------------------------------------------------------------ + # Event queue (register + long-poll) + # ------------------------------------------------------------------ + + async def _register_queue(self) -> bool: + all_public = os.getenv("ZULIP_ALL_PUBLIC_STREAMS", "").lower() in {"1", "true", "yes", "on"} + payload = { + "event_types": json.dumps(["message"]), + "apply_markdown": "false", + "all_public_streams": "true" if all_public else "false", + } + data = await self._api_post("register", payload) + if data.get("result") != "success" or "queue_id" not in data: + logger.error("Zulip: failed to register event queue: %s", str(data.get("msg", ""))[:160]) + return False + self._queue_id = data["queue_id"] + self._last_event_id = data.get("last_event_id", -1) + logger.info("Zulip: event queue registered (%s)", self._queue_id) + return True + + async def _poll_loop(self) -> None: + delay = _RECONNECT_BASE_DELAY + while not self._closing: + try: + if not self._queue_id: + if not await self._register_queue(): + raise RuntimeError("queue registration failed") + await self._poll_once() + delay = _RECONNECT_BASE_DELAY + except asyncio.CancelledError: + return + except Exception as exc: + if self._closing: + return + logger.warning("Zulip poll error: %s — retrying in %.0fs", exc, delay) + import random + await asyncio.sleep(delay + delay * _RECONNECT_JITTER * random.random()) + delay = min(delay * 2, _RECONNECT_MAX_DELAY) + + async def _get_events(self) -> Optional[Dict[str, Any]]: + """Long-poll ``GET /events``. + + Returns the parsed response, or ``None`` on the expected long-poll + timeout (no events within the window — caller just re-polls). Kept + separate from ``_api_get`` so the long-poll timeout isn't swallowed + into an error dict the way ordinary calls need. + """ + import aiohttp + params = {"queue_id": self._queue_id, "last_event_id": self._last_event_id} + try: + async with self._session.get( + f"{self._site}/api/v1/events", params=params, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=_LONGPOLL_TIMEOUT), + ) as resp: + return await resp.json() + except asyncio.TimeoutError: + return None + except aiohttp.ClientError as exc: + raise RuntimeError(f"events poll network error: {exc}") + + async def _poll_once(self) -> None: + data = await self._get_events() + if data is None: + return # long-poll window elapsed — just poll again + + if data.get("result") != "success": + if data.get("code") == "BAD_EVENT_QUEUE_ID": + logger.info("Zulip: event queue expired — re-registering") + self._queue_id = None + return + # Transient server error — let the loop back off. + raise RuntimeError(str(data.get("msg", "events poll failed"))) + + for event in data.get("events", []): + eid = event.get("id") + if isinstance(eid, int): + self._last_event_id = max(self._last_event_id, eid) + if event.get("type") == "message": + # Guard each event so one failing message can't drop the rest + # of the batch. last_event_id is already advanced above, so a + # dropped message is not reprocessed on the next poll. + try: + await self._handle_message_event(event) + except Exception: + logger.exception("Zulip: error handling message event %s", eid) + + async def _handle_message_event(self, event: Dict[str, Any]) -> None: + message = event.get("message") or {} + flags = event.get("flags") or [] + + sender_id = message.get("sender_id") + # Ignore our own messages (Zulip echoes them back on the queue). + if sender_id is not None and sender_id == self._bot_user_id: + return + + msg_id = message.get("id") + if msg_id is None or self._dedup.is_duplicate(str(msg_id)): + return + + mtype = message.get("type") # "stream" or "private" + content = message.get("content", "") or "" + sender_name = message.get("sender_full_name", "") or message.get("sender_email", "") + + if mtype == "private": + chat_type = "dm" + recipients = message.get("display_recipient") or [] + user_ids = sorted( + int(r["id"]) for r in recipients + if isinstance(r, dict) and "id" in r + ) + chat_id = "d:" + ",".join(str(u) for u in user_ids) + topic = None + else: + chat_type = "channel" + stream_id = message.get("stream_id") + topic = message.get("subject", "") or message.get("topic", "") + chat_id = f"s:{stream_id}:{topic}" + + # Mention-gating for streams. + stream_name = message.get("display_recipient", "") + extra = self.config.extra or {} + + allowed_raw = extra.get("allowed_channels") + if allowed_raw is None: + allowed_raw = os.getenv("ZULIP_ALLOWED_CHANNELS", "") + allowed = _split_csv(allowed_raw) + if allowed and str(stream_id) not in allowed and str(stream_name) not in allowed: + return + + require_mention = os.getenv("ZULIP_REQUIRE_MENTION", "true").lower() not in {"false", "0", "no"} + free = _split_csv(os.getenv("ZULIP_FREE_RESPONSE_CHANNELS", "")) + is_free = str(stream_id) in free or str(stream_name) in free + + has_mention = "mentioned" in flags or "wildcard_mentioned" in flags + if require_mention and not is_free and not has_mention: + return + + # Strip the bot's @-mention so the agent sees clean input. + if self._bot_full_name: + content = re.sub( + r"@\*\*" + re.escape(self._bot_full_name) + r"\*\*", "", content, + ).strip() + + msg_type = MessageType.COMMAND if content.startswith("/") else MessageType.TEXT + + media_urls: List[str] = [] + media_types: List[str] = [] + await self._extract_inline_media(content, media_urls, media_types) + await self._extract_event_attachments(message, media_urls, media_types) + if media_types and msg_type == MessageType.TEXT: + if any(m.startswith("image/") for m in media_types): + msg_type = MessageType.PHOTO + elif any(m.startswith("audio/") for m in media_types): + msg_type = MessageType.VOICE + else: + msg_type = MessageType.DOCUMENT + + source = self.build_source( + chat_id=chat_id, + chat_type=chat_type, + user_id=str(sender_id) if sender_id is not None else None, + user_name=sender_name, + chat_topic=topic, + ) + + from gateway.platforms.base import resolve_channel_prompt + channel_prompt = resolve_channel_prompt(self.config.extra, chat_id, None) + + await self.handle_message(MessageEvent( + text=content, + message_type=msg_type, + source=source, + raw_message=message, + message_id=str(msg_id), + media_urls=media_urls or None, + media_types=media_types or None, + channel_prompt=channel_prompt, + )) + + async def _extract_event_attachments( + self, message: Dict[str, Any], media_urls: List[str], media_types: List[str], + ) -> None: + """Download files from Zulip's ``message.attachments`` field (UI file + uploads that Zulip sends as structured event data rather than inline + markdown links).""" + import aiohttp + atts = message.get("attachments") or [] + for att in atts: + path_id = att.get("path_id") + fname = att.get("name") or "file" + if not path_id: + continue + dl_url = f"{self._site}/api/v1/user_uploads/{path_id}" + try: + async with self._session.get( + dl_url, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status >= 400: + logger.debug("Zulip: event attachment %s -> %s", fname, resp.status) + continue + data = await resp.read() + mime = resp.content_type or "application/octet-stream" + except (aiohttp.ClientError, asyncio.TimeoutError): + logger.debug("Zulip: failed to download event attachment %s", fname) + continue + await self._cache_attachment(data, fname, mime, media_urls, media_types) + + async def _cache_attachment( + self, data: bytes, fname: str, mime: str, + media_urls: List[str], media_types: List[str], + ) -> None: + """Cache a downloaded attachment into Hermes media storage.""" + from gateway.platforms.base import ( + cache_image_from_bytes, cache_audio_from_bytes, cache_document_from_bytes, + ) + ext = Path(fname).suffix + try: + if mime.startswith("image/"): + media_urls.append(cache_image_from_bytes(data, ext or ".png")) + media_types.append(mime) + elif mime.startswith("audio/"): + media_urls.append(cache_audio_from_bytes(data, ext or ".ogg")) + media_types.append(mime) + else: + media_urls.append(cache_document_from_bytes(data, fname)) + media_types.append(mime) + except Exception as exc: + logger.warning("Zulip: failed to cache attachment %s: %s", fname, exc) + + async def _extract_inline_media( + self, content: str, media_urls: List[str], media_types: List[str], + ) -> None: + """Download Zulip ``/user_uploads/...`` attachments referenced in a + message so vision/transcription tools can read them locally.""" + import aiohttp + for rel in re.findall(r"\]\((/user_uploads/[^)]+)\)", content): + dl_url = f"{self._site}{rel}" + try: + async with self._session.get( + dl_url, auth=self._auth(), + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status >= 400: + continue + data = await resp.read() + mime = resp.content_type or "application/octet-stream" + except (aiohttp.ClientError, asyncio.TimeoutError): + continue + fname = rel.rsplit("/", 1)[-1] or "file" + await self._cache_attachment(data, fname, mime, media_urls, media_types) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _split_csv(value: Any) -> set: + if isinstance(value, list): + return {str(v).strip() for v in value if str(v).strip()} + return {v.strip() for v in str(value or "").split(",") if v.strip()} + + +# --------------------------------------------------------------------------- +# Out-of-process cron delivery (standalone REST send) +# --------------------------------------------------------------------------- + + +async def _standalone_send( + pconfig, + chat_id: str, + message: str, + *, + thread_id: Optional[str] = None, + media_files: Optional[list] = None, + force_document: bool = False, +) -> Dict[str, Any]: + """Send via the Zulip REST API without a live gateway adapter. + + Used by ``tools/send_message_tool`` for cron jobs running outside the + gateway process. Reads ``ZULIP_API_KEY`` from ``pconfig.token`` (env + fallback); site + email come from ``pconfig.extra`` or env. + """ + try: + import aiohttp + except ImportError: + return {"error": "aiohttp not installed. Run: pip install aiohttp"} + + extra = getattr(pconfig, "extra", {}) or {} + site = (extra.get("site") or os.getenv("ZULIP_SITE", "") or os.getenv("ZULIP_URL", "")).rstrip("/") + email = (extra.get("email") or os.getenv("ZULIP_EMAIL", "")).strip() + api_key = (getattr(pconfig, "token", None) or os.getenv("ZULIP_API_KEY", "")).strip() + if not site or not email or not api_key: + return {"error": "Zulip standalone send: ZULIP_SITE, ZULIP_EMAIL and ZULIP_API_KEY must be set"} + + default_topic = extra.get("default_topic") or os.getenv("ZULIP_DEFAULT_TOPIC", "general") + kind, to, topic = _parse_target(chat_id, default_topic) + + try: + from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_aiohttp + proxy = resolve_proxy_url(platform_env_var="ZULIP_PROXY") + sess_kw, req_kw = proxy_kwargs_for_aiohttp(proxy) + auth = aiohttp.BasicAuth(email, api_key) + + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60), **sess_kw) as session: + body = message + + # Upload any media and append links to the message body. + for media in (media_files or []): + file_path = media.get("path") if isinstance(media, dict) else media + if not file_path or not os.path.exists(file_path): + continue + form = aiohttp.FormData() + with open(file_path, "rb") as fh: + form.add_field("file", fh.read(), filename=os.path.basename(file_path)) + async with session.post( + f"{site}/api/v1/user_uploads", data=form, auth=auth, **req_kw, + ) as up: + if up.status >= 400: + continue + up_data = await up.json() + rel = up_data.get("url") or up_data.get("uri") + if rel: + full = rel if rel.startswith("http") else f"{site}{rel}" + body = f"{body}\n[{os.path.basename(file_path)}]({full})".strip() + + payload: Dict[str, Any] = {"content": body or "(no content)"} + if kind == "direct": + # "private" is the backward-compatible alias accepted by every + # Zulip version (the newer "direct" alias only landed in 7.0). + payload["type"] = "private" + payload["to"] = json.dumps(to) + else: + payload["type"] = "stream" + payload["to"] = str(to) + payload["topic"] = topic or default_topic + + async with session.post( + f"{site}/api/v1/messages", data=payload, auth=auth, **req_kw, + ) as resp: + data = await resp.json() + if data.get("result") != "success": + return {"error": f"Zulip API error: {str(data.get('msg', ''))[:300]}"} + return { + "success": True, + "platform": "zulip", + "chat_id": chat_id, + "message_id": data.get("id"), + } + except aiohttp.ClientError as exc: + return {"error": f"Zulip send failed (network): {exc}"} + except Exception as exc: # noqa: BLE001 + return {"error": f"Zulip send failed: {exc}"} + + +# --------------------------------------------------------------------------- +# Interactive setup wizard +# --------------------------------------------------------------------------- + + +def interactive_setup() -> None: + """Guide the user through Zulip bot setup.""" + from hermes_cli.config import get_env_value, save_env_value + from hermes_cli.cli_output import ( + prompt, prompt_yes_no, print_header, print_info, print_success, + ) + + print_header("Zulip") + if get_env_value("ZULIP_API_KEY") and get_env_value("ZULIP_EMAIL"): + print_info("Zulip: already configured") + if not prompt_yes_no("Reconfigure Zulip?", False): + return + + print_info("Create a bot in Zulip: Settings → Personal → Bots → Add a new bot") + print_info(" Choose 'Generic bot', then copy its email + API key.") + print() + site = prompt("Zulip site URL (e.g. https://example.zulipchat.com)") + if site: + save_env_value("ZULIP_SITE", site.rstrip("/")) + email = prompt("Bot email") + if email: + save_env_value("ZULIP_EMAIL", email) + api_key = prompt("Bot API key", password=True) + if not api_key: + return + save_env_value("ZULIP_API_KEY", api_key) + print_success("Zulip credentials saved") + + print() + print_info("🔒 Security: restrict who can use your bot") + allowed = prompt("Allowed sender emails/ids (comma-separated, empty for open access)") + if allowed: + save_env_value("ZULIP_ALLOWED_USERS", allowed.replace(" ", "")) + print_success("Zulip allowlist configured") + else: + print_info("⚠️ No allowlist set — anyone who can message the bot can use it!") + + print() + print_info("📬 Home channel: where Hermes delivers cron results / notifications.") + print_info(" Form: 'stream-name/topic' (e.g. general/hermes).") + home = prompt("Home channel (leave empty to set later with /set-home)") + if home: + save_env_value("ZULIP_HOME_CHANNEL", home) + + +# --------------------------------------------------------------------------- +# YAML -> env config bridge / connected probe +# --------------------------------------------------------------------------- + + +def _apply_yaml_config(yaml_cfg: dict, z_cfg: dict) -> dict | None: + extras: Dict[str, Any] = {} + site = z_cfg.get("site") or z_cfg.get("url") + if site: + extras["site"] = str(site).rstrip("/") + if not os.getenv("ZULIP_SITE"): + os.environ["ZULIP_SITE"] = str(site).rstrip("/") + if z_cfg.get("email"): + extras["email"] = str(z_cfg["email"]) + if not os.getenv("ZULIP_EMAIL"): + os.environ["ZULIP_EMAIL"] = str(z_cfg["email"]) + if z_cfg.get("default_topic"): + extras["default_topic"] = str(z_cfg["default_topic"]) + if "require_mention" in z_cfg and not os.getenv("ZULIP_REQUIRE_MENTION"): + os.environ["ZULIP_REQUIRE_MENTION"] = str(z_cfg["require_mention"]).lower() + if "all_public_streams" in z_cfg and not os.getenv("ZULIP_ALL_PUBLIC_STREAMS"): + os.environ["ZULIP_ALL_PUBLIC_STREAMS"] = str(z_cfg["all_public_streams"]).lower() + for key, env in ( + ("free_response_channels", "ZULIP_FREE_RESPONSE_CHANNELS"), + ("allowed_channels", "ZULIP_ALLOWED_CHANNELS"), + ): + val = z_cfg.get(key) + if val is not None and not os.getenv(env): + if isinstance(val, list): + val = ",".join(str(v) for v in val) + os.environ[env] = str(val) + return extras or None + + +def _env_enablement() -> Optional[dict]: + site = os.getenv("ZULIP_SITE", "") or os.getenv("ZULIP_URL", "") + if not site: + return None + extras: Dict[str, Any] = {"site": site.rstrip("/")} + if os.getenv("ZULIP_EMAIL"): + extras["email"] = os.getenv("ZULIP_EMAIL") + return extras + + +def _is_connected(config) -> bool: + import hermes_cli.gateway as gateway_mod + site = (gateway_mod.get_env_value("ZULIP_SITE") or gateway_mod.get_env_value("ZULIP_URL") or "").strip() + email = (gateway_mod.get_env_value("ZULIP_EMAIL") or "").strip() + api_key = (gateway_mod.get_env_value("ZULIP_API_KEY") or "").strip() + return bool(site and email and api_key) + + +# --------------------------------------------------------------------------- +# Plugin registration entry point +# --------------------------------------------------------------------------- + + +def _build_adapter(config): + return ZulipAdapter(config) + + +def register(ctx) -> None: + """Plugin entry point — called by the Hermes plugin system.""" + ctx.register_platform( + name=PLATFORM_NAME, + label="Zulip", + adapter_factory=_build_adapter, + check_fn=check_zulip_requirements, + is_connected=_is_connected, + required_env=["ZULIP_SITE", "ZULIP_EMAIL", "ZULIP_API_KEY"], + install_hint="pip install aiohttp", + setup_fn=interactive_setup, + apply_yaml_config_fn=_apply_yaml_config, + env_enablement_fn=_env_enablement, + allowed_users_env="ZULIP_ALLOWED_USERS", + allow_all_env="ZULIP_ALLOW_ALL_USERS", + cron_deliver_env_var="ZULIP_HOME_CHANNEL", + standalone_sender_fn=_standalone_send, + max_message_length=MAX_MESSAGE_LENGTH, + emoji="🗨️", + platform_hint=( + "You are on Zulip. Messages live under a channel (stream) + topic, " + "or are direct messages. Use CommonMark/Zulip Markdown for formatting " + "(**bold**, *italic*, `code`, ```code blocks```, lists, links, " + "spoilers, LaTeX). Keep replies on-topic; the gateway keeps your " + "reply in the same stream/topic or DM it arrived on." + ), + allow_update_command=True, + ) diff --git a/pi-zulip-extension/config.yaml.example b/pi-zulip-extension/config.yaml.example new file mode 100644 index 0000000..0a0a75b --- /dev/null +++ b/pi-zulip-extension/config.yaml.example @@ -0,0 +1,24 @@ +# pi Zulip extension — Abiba configuration (flat format for custom parser) +# Deploy to: ~/.pi/agent/extensions/config.yaml + +zulip.site: https://chat.sysloggh.net +zulip.email: abiba-bot@chat.sysloggh.net +zulip.api_key: cKTDMZAPW08dk3zl05sStzO7HRztzyn8 +zulip.stream: agent-hub +zulip.all_bots_user_id: 20 + +agent.name: abiba +agent.display_name: Abiba +agent.zulip_bot_name: abiba-bot +agent.owner_email: jerome@sysloggh.com +agent.private_topic: abiba + +health_port: 9200 + +monitoring.health_endpoint_enabled: true +monitoring.log_level: info + +error_handling.timeout_seconds: 30 +error_handling.retry_count: 3 +error_handling.retry_delay_seconds: 5 +error_handling.graceful_message: "Abiba encountered an error processing your request. Please try again later." diff --git a/pi-zulip-extension/config/ecosystem.abiba.config.cjs b/pi-zulip-extension/config/ecosystem.abiba.config.cjs new file mode 100644 index 0000000..74eacc2 --- /dev/null +++ b/pi-zulip-extension/config/ecosystem.abiba.config.cjs @@ -0,0 +1,26 @@ +module.exports = { + apps: [ + { + name: "abiba-zulip", + script: "/usr/bin/pi", + args: "--mode rpc --session-id zulip-service", + cwd: "/root", + env: { + ZULIP_ROLE: "router", + ZULIP_EXTENSION_ACTIVE: "true", + NODE_OPTIONS: "--max-old-space-size=512", + }, + // Prevent rapid crash-looping: restart with backoff, limit retries + min_uptime: "30s", + max_restarts: 15, + restart_delay: 10000, + kill_timeout: 5000, + autorestart: true, + }, + { + name: "abiba-telegram", + script: "/usr/bin/pitg", + cwd: "/root", + }, + ], +}; diff --git a/pi-zulip-extension/src/index.d.ts b/pi-zulip-extension/src/index.d.ts new file mode 100644 index 0000000..b46eb6f --- /dev/null +++ b/pi-zulip-extension/src/index.d.ts @@ -0,0 +1,16 @@ +/** + * pi-zulip-extension — Zulip agent communication plugin for pi agents + * + * Deploy to: ~/.pi/agent/extensions/zulip/ + * Config: config.yaml (alongside extension, or env vars) + * + * DM-first architecture per ADR-001/ADR-002. + * Background Zulip event queue poller that injects messages directly into + * the current pi session via pi.sendUserMessage(), then captures the LLM + * response from agent_end and sends it back to Zulip. No subprocess overhead. + * + * @see ADR-007: Platform-native plugin contracts + * @see docs/ARCHITECTURE.md + */ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; +export default function (pi: ExtensionAPI): void; diff --git a/pi-zulip-extension/src/index.js b/pi-zulip-extension/src/index.js new file mode 100644 index 0000000..b3d5de1 --- /dev/null +++ b/pi-zulip-extension/src/index.js @@ -0,0 +1,1255 @@ +/** + * pi-zulip-extension v2 — Router-worker architecture for per-sender session isolation + * + * Deploy to: ~/.pi/agent/extensions/zulip/ + * Config: config.yaml (alongside extension) + * + * Architecture: + * Router (this process, PM2 abiba-zulip, ZULIP_ROLE=router): + * Polls Zulip for events, maintains a pool of per-sender pi RPC workers. + * Each sender gets their own pi --mode rpc process with a dedicated session file. + * Messages are routed to the correct worker; worker responses are relayed + * back to Zulip (with streaming edits). + * Worker (child pi --mode rpc --session-dir ...): + * Extension loads but is NO-OP (ZULIP_ROLE != router). Just runs the agent + * with per-sender persistent sessions. No Zulip logic in the worker. + */ + +import fs from "node:fs"; +import path from "node:path"; +import url from "node:url"; +import http from "node:http"; +import { spawn } from "node:child_process"; +import { execSync } from "node:child_process"; +import { parse } from "yaml"; + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- +const EXT_DIR = path.dirname(url.fileURLToPath(import.meta.url)); +const SESSION_DIR = path.join(process.env.HOME, ".pi", "agent", "sessions", "zulip"); +const SESSIONS_FILE = path.join(SESSION_DIR, "worker-sessions.json"); +const POLL_INTERVAL_MS = 3000; +const WORKER_IDLE_TIMEOUT_MS = 30 * 60 * 1000; // kill idle workers after 30 min +const STREAMING_EDIT_MS = 8000; // throttle Zulip message edits during streaming +const MAX_ZULIP_MSG = 10000; +const HEARTBEAT_INTERVAL_MS = 5 * 60 * 1000; + +// --------------------------------------------------------------------------- +// Config loading (cleaned — flat yaml format expected) +// --------------------------------------------------------------------------- +function loadConfig() { + const configPath = path.join(EXT_DIR, "config.yaml"); + if (!fs.existsSync(configPath)) { + throw new Error(`No config.yaml at ${configPath}`); + } + const raw = fs.readFileSync(configPath, "utf-8"); + const cfg = parse(raw); + + function get(key) { + if (cfg[key] !== undefined) return cfg[key]; + const parts = key.split("."); + let cur = cfg; + for (const p of parts) { + if (cur && typeof cur === "object" && p in cur) cur = cur[p]; + else return undefined; + } + return cur; + } + + return { + zulip: { + email: String(get("zulip.email") ?? ""), + api_key: process.env.ZULIP_API_KEY ?? String(get("zulip.api_key") ?? ""), + site: String(get("zulip.site") ?? ""), + all_bots_user_id: Number(get("zulip.all_bots_user_id") ?? 1), + }, + agent: { + name: String(get("agent.name") ?? "abiba"), + owner_email: String(get("agent.owner_email") ?? "jerome@sysloggh.com"), + }, + health_port: parseInt(String(get("health_port") ?? "9200"), 10), + }; +} + +const config = loadConfig(); + +// --------------------------------------------------------------------------- +// Zulip client (raw fetch — no zulip-js dependency) +// --------------------------------------------------------------------------- +function authHeader() { + return "Basic " + Buffer.from(`${config.zulip.email}:${config.zulip.api_key}`).toString("base64"); +} + +/** + * Create a Zulip event queue and resolve bot + all-bots user IDs. + */ +async function createZulipQueue() { + const auth = authHeader(); + + // Register event queue + const regForm = new URLSearchParams(); + regForm.append("event_types", JSON.stringify(["message"])); + const regRes = await fetch(`${config.zulip.site}/api/v1/register`, { + method: "POST", + headers: { Authorization: auth, "Content-Type": "application/x-www-form-urlencoded" }, + body: regForm, + }); + if (!regRes.ok) throw new Error(`Queue registration failed: ${regRes.status} ${await regRes.text().catch(() => '').then(t => t.slice(0,100))}`); + const qr = await regRes.json(); + + let lastEventId = qr.last_event_id ?? -1; + let botUserId = null; + let allBotsUserId = config.zulip.all_bots_user_id ?? 1; + + // Resolve bot's own user_id + try { + const meRes = await fetch(`${config.zulip.site}/api/v1/users/me`, { headers: { Authorization: auth } }); + if (meRes.ok) { + const md = await meRes.json(); + if (md.user_id) botUserId = md.user_id; + } + } catch { /* non-critical */ } + + // Resolve @all-bots user_id dynamically + try { + const usersRes = await fetch(`${config.zulip.site}/api/v1/users`, { headers: { Authorization: auth } }); + if (usersRes.ok) { + const ud = await usersRes.json(); + for (const m of ud.members || []) { + if ((m.email || "").toLowerCase().includes("all-bots")) { + allBotsUserId = m.user_id; + break; + } + } + } + } catch { /* fall back to config */ } + + console.log(`[zulip-ext] Bot user_id=${botUserId}, all-bots user_id=${allBotsUserId}`); + + return { + queueId: qr.queue_id, + lastEventId, + botUserId, + allBotsUserId, + async poll() { + const res = await fetch( + `${config.zulip.site}/api/v1/events?queue_id=${encodeURIComponent(qr.queue_id)}&last_event_id=${lastEventId}`, + { headers: { Authorization: authHeader() } } + ); + if (!res.ok) { + if (res.status === 429) { + const ra = parseFloat(res.headers.get("retry-after") || "1"); + await new Promise(r => setTimeout(r, ra * 1000 + 100)); + return []; + } + const errBody = await res.text().catch(() => ''); + throw new Error(`Events API ${res.status}: ${errBody.slice(0, 200)}`); + } + const data = await res.json(); + if (data.events) { + for (const e of data.events) { + if (e.id > lastEventId) lastEventId = e.id; + } + return data.events.filter(e => e.type === "message"); + } + return []; + }, + async sendMessage(params) { + const fb = new URLSearchParams(); + fb.append("type", params.type); + fb.append("to", params.type === "private" ? JSON.stringify([params.to]) : String(params.to)); + if (params.subject) fb.append("subject", params.subject); + fb.append("content", params.content); + const res = await fetch(`${config.zulip.site}/api/v1/messages`, { + method: "POST", + headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" }, + body: fb.toString(), + }); + if (!res.ok) { const errBody = await res.text().catch(() => ''); throw new Error(`Send failed ${res.status}: ${(errBody || '').slice(0, 200)}`); } + return (await res.json()).id; + }, + async editMessage(messageId, content) { + const fb = new URLSearchParams(); + fb.append("content", content); + const res = await fetch(`${config.zulip.site}/api/v1/messages/${messageId}`, { + method: "PATCH", + headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" }, + body: fb.toString(), + }); + if (!res.ok) { + console.warn(`[zulip-ext] Edit ${messageId} failed: ${res.status}`); + } + }, + async sendTyping(userIds, op) { + const fb = new URLSearchParams(); + fb.append("to", JSON.stringify(userIds)); + fb.append("op", op); + await fetch(`${config.zulip.site}/api/v1/typing`, { + method: "POST", + headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" }, + body: fb.toString(), + }).catch(() => {}); + }, + }; +} + +// --------------------------------------------------------------------------- +// Dynamic echo prevention — fetch bot users from Zulip API +// --------------------------------------------------------------------------- +let BOT_EMAILS = new Set([config.zulip.email]); // always include self + +async function loadBotEmails(queue) { + try { + const res = await fetch(`${config.zulip.site}/api/v1/users`, { + headers: { Authorization: authHeader() }, + }); + if (res.ok) { + const data = await res.json(); + const count = BOT_EMAILS.size; + for (const m of data.members || []) { + if (m.is_bot && m.email) BOT_EMAILS.add(m.email); + } + console.log(`[zulip-ext] Echo prevention: ${BOT_EMAILS.size} bot emails (${BOT_EMAILS.size - count} dynamic)`); + } + } catch (e) { + console.warn(`[zulip-ext] Could not fetch bot list: ${e.message}`); + } +} + +function isBotSender(email) { + return BOT_EMAILS.has(email); +} + +const ALL_BOTS_RE = /@\*\*(?:all-bots|All Bots)\*\*/i; +function isAllBotsMention(content) { + return ALL_BOTS_RE.test(content); +} + +// --------------------------------------------------------------------------- +// Worker pool manager +// --------------------------------------------------------------------------- +/** @type {Map} */ +const workers = new Map(); +let workerSessions = {}; // senderId → sessionFilePath (persisted to disk) +let cmdIdCounter = 0; +let idleCleanupTimer = null; + +function loadWorkerSessions() { + try { + if (fs.existsSync(SESSIONS_FILE)) { + return JSON.parse(fs.readFileSync(SESSIONS_FILE, "utf-8")); + } + } catch (e) { + console.warn(`[zulip-ext] Failed to load worker sessions: ${e.message}`); + } + return {}; +} + +function saveWorkerSessions() { + try { + if (!fs.existsSync(SESSION_DIR)) fs.mkdirSync(SESSION_DIR, { recursive: true }); + fs.writeFileSync(SESSIONS_FILE, JSON.stringify(workerSessions, null, 2)); + } catch (e) { + console.warn(`[zulip-ext] Failed to save worker sessions: ${e.message}`); + } +} + +/** + * Attach a JSONL line reader to a stream. + */ +function attachJsonlReader(stream, onLine, onError) { + let buf = ""; + const td = new TextDecoder(); + stream.on("data", (chunk) => { + buf += typeof chunk === "string" ? chunk : td.decode(chunk, { stream: true }); + while (true) { + const idx = buf.indexOf("\n"); + if (idx === -1) break; + const line = buf.slice(0, idx).replace(/\r$/, ""); + buf = buf.slice(idx + 1); + if (!line) continue; + try { + onLine(JSON.parse(line)); + } catch (e) { + if (onError) onError(e, line); + } + } + }); +} + +/** + * Spawn a pi RPC worker process for a sender. + */ +function spawnWorker(senderId) { + const sessionDir = SESSION_DIR; + if (!fs.existsSync(sessionDir)) fs.mkdirSync(sessionDir, { recursive: true }); + + const proc = spawn("pi", [ + "--mode", "rpc", + "--session-dir", sessionDir, + ], { + env: { + ...process.env, + ZULIP_ROLE: "worker", + ZULIP_EXTENSION_ACTIVE: "", // ensure NOT active + }, + stdio: ["pipe", "pipe", "pipe"], + cwd: process.env.HOME, + }); + + const worker = { + process: proc, + senderId, + busy: false, + streamingBuffer: "", + lastStreamEdit: 0, + pendingReplies: [], // [{ zulipMsgId, senderId, senderName, replyInfo, createdAt }] + lastMessageContent: null, + lastReplyInfo: null, + lastActivity: Date.now(), + createdAt: Date.now(), + sessionFile: workerSessions[senderId] || null, + pendingCommands: new Map(), + rpcCommandQueue: [], + }; + + // Read stdout events (JSONL) + attachJsonlReader( + proc.stdout, + (event) => handleWorkerEvent(worker, event), + (err, line) => console.warn(`[zulip-ext] Worker ${senderId} parse error: ${err.message} | ${line.slice(0, 100)}`) + ); + + // Pipe stderr for diagnostics + proc.stderr.on("data", (d) => { + const text = d.toString().trim(); + if (text) process.stderr.write(`[worker-${senderId}] ${text}\n`); + }); + + // Handle exit + proc.on("exit", (code, signal) => { + console.log(`[zulip-ext] Worker ${senderId} exited: code=${code} signal=${signal || "none"}`); + // Fail any pending replies + while (worker.pendingReplies.length > 0) { + const reply = worker.pendingReplies.shift(); + if (reply.zulipMsgId && zulipQueue) { + zulipQueue.editMessage(reply.zulipMsgId, + ":warning: Response interrupted — worker process exited. Please try again." + ).catch(() => {}); + } + } + workers.delete(senderId); + }); + + // Resolve session file + (async () => { + if (worker.sessionFile) { + // Resume existing session + const cmdId = String(++cmdIdCounter); + sendRpc(worker, { id: cmdId, type: "switch_session", sessionPath: worker.sessionFile }); + console.log(`[zulip-ext] Worker ${senderId}: resuming session ${worker.sessionFile}`); + } + // If no session file, the auto-created session is fine. Discover it. + if (!worker.sessionFile) { + const cmdId = String(++cmdIdCounter); + sendRpc(worker, { id: cmdId, type: "get_state" }); + } + })(); + + return worker; +} + +function sendRpc(worker, cmd) { + worker.process.stdin.write(JSON.stringify(cmd) + "\n"); +} + +/** + * Handle all RPC events from a worker. + */ +function handleWorkerEvent(worker, event) { + // Debug: log all worker events to diagnose processing + if (event.type !== 'message_update' || !event.assistantMessageEvent || event.assistantMessageEvent.type === 'text_start' || event.assistantMessageEvent.type === 'text_end') { + console.log(`[zulip-ext] Worker ${worker.senderId} event: ${event.type}${event.assistantMessageEvent ? ':' + event.assistantMessageEvent.type : ''}${event.command ? ' cmd=' + event.command : ''}`); + } + switch (event.type) { + case "response": + handleWorkerResponse(worker, event); + break; + case "message_update": + handleWorkerStreaming(worker, event); + break; + case "agent_end": + handleWorkerAgentEnd(worker, event); + break; + case "agent_start": + break; + case "turn_start": + break; + case "turn_end": + break; + case "compaction_start": + console.log(`[zulip-ext] Worker ${worker.senderId}: compaction started`); + break; + case "compaction_end": + console.log(`[zulip-ext] Worker ${worker.senderId}: compaction ended`); + // Track new session file path if it changed + break; + case "extension_error": + console.error(`[zulip-ext] Worker ${worker.senderId} ext error: ${event.error?.slice(0, 200)}`); + break; + case "message_start": + case "message_end": + case "tool_execution_start": + case "tool_execution_update": + case "tool_execution_end": + case "queue_update": + case "auto_retry_start": + case "auto_retry_end": + break; // informational, not needed by router + } +} + +function handleWorkerResponse(worker, event) { + // Resolve pending command callbacks + if (event.id) { + const cb = worker.pendingCommands.get(event.id); + if (cb) { + worker.pendingCommands.delete(event.id); + cb.resolve(event); + return; + } + } + + // Handle responses without explicit callbacks + if (event.command === "get_state" && event.success) { + const sessionFile = event.data?.sessionFile; + if (sessionFile && !worker.sessionFile) { + worker.sessionFile = sessionFile; + workerSessions[worker.senderId] = sessionFile; + saveWorkerSessions(); + console.log(`[zulip-ext] Worker ${worker.senderId}: session file = ${sessionFile}`); + } + } + + if (event.command === "switch_session" && event.success) { + console.log(`[zulip-ext] Worker ${worker.senderId}: session switched, cancelled=${event.data?.cancelled}`); + } + + if (event.command === "new_session" && event.success) { + // Track the new session file (get_state will follow if needed) + if (!event.data?.cancelled) { + const cmdId = String(++cmdIdCounter); + worker.pendingCommands.set(cmdId, { + resolve: (ev) => { + const sf = ev.data?.sessionFile; + if (sf) { + worker.sessionFile = sf; + workerSessions[worker.senderId] = sf; + saveWorkerSessions(); + console.log(`[zulip-ext] Worker ${worker.senderId}: new session = ${sf}`); + } + }, + }); + sendRpc(worker, { id: cmdId, type: "get_state" }); + } + } +} + +function handleWorkerStreaming(worker, event) { + const delta = event.assistantMessageEvent; + if (!delta || delta.type !== "text_delta") return; + + worker.streamingBuffer += delta.delta; + const now = Date.now(); + if (worker.streamingBuffer.length > 10 && (!worker.lastStreamEdit || now - worker.lastStreamEdit >= STREAMING_EDIT_MS)) { + worker.lastStreamEdit = now; + const reply = worker.pendingReplies[0]; + if (reply?.zulipMsgId && zulipQueue) { + const preview = worker.streamingBuffer.length > 9500 + ? worker.streamingBuffer.slice(0, 9500) + "\n\n_… still generating…_" + : worker.streamingBuffer; + zulipQueue.editMessage(reply.zulipMsgId, preview).catch(() => {}); + } + } +} + +async function handleWorkerAgentEnd(worker, event) { + worker.busy = false; + worker.streamingBuffer = ""; + worker.lastStreamEdit = 0; + worker.lastActivity = Date.now(); + + if (worker.pendingReplies.length === 0) return; + const reply = worker.pendingReplies.shift(); + + // Extract final assistant text — walk backwards to find the last message with text content + // (the final turn may only have tool calls with brief inline text like "On it. Let me") + const assistantMsgs = (event.messages || []).filter(m => m.role === "assistant"); + let responseText = ""; + for (let i = assistantMsgs.length - 1; i >= 0; i--) { + const content = assistantMsgs[i].content; + if (!content) continue; + if (typeof content === "string") { + responseText = content; + break; + } + if (Array.isArray(content)) { + const texts = content.filter(c => c.type === "text").map(c => c.text).join("\n"); + if (texts.trim()) { + responseText = texts; + break; + } + } + } + + // Stop typing + if (reply.senderId && zulipQueue) { + zulipQueue.sendTyping([reply.senderId], "stop").catch(() => {}); + } + + if (!responseText.trim()) { + // All assistant messages are tool-call-only. Send a summary. + console.log(`[zulip-ext] Tool-only response for ${reply.senderName} — sending activity note`); + responseText = "⚙️ _Running tools…_\n\nUse `/continue` to see results or send another message."; + } + + const truncated = responseText.length > MAX_ZULIP_MSG + ? responseText.slice(0, MAX_ZULIP_MSG) + "\n\n[...truncated at Zulip limit]" + : responseText; + + const target = reply.replyInfo.type === "private" + ? { type: "private", to: reply.senderId } + : { type: "stream", to: reply.replyInfo.streamId ?? reply.replyInfo.streamName, subject: reply.replyInfo.topic }; + + try { + if (reply.zulipMsgId && zulipQueue) { + await zulipQueue.editMessage(reply.zulipMsgId, truncated); + console.log(`[zulip-ext] Finalized reply for ${reply.senderName} on worker ${worker.senderId}`); + } else if (zulipQueue) { + await zulipQueue.sendMessage({ ...target, content: truncated }); + console.log(`[zulip-ext] Sent reply for ${reply.senderName}`); + } + } catch (err) { + console.error(`[zulip-ext] Failed to send reply: ${err.message}`); + } +} + +/** + * Kill a worker and clean up. + */ +function killWorker(senderId) { + const w = workers.get(senderId); + if (!w) return; + console.log(`[zulip-ext] Killing idle worker ${senderId}`); + workers.delete(senderId); + try { w.process.kill("SIGTERM"); } catch { /* already dead */ } + // Note: workerSessions entry is kept so session resumes on re-connect +} + +/** + * Periodically kill idle workers. + */ +function cleanupIdleWorkers() { + const now = Date.now(); + for (const [senderId, w] of workers) { + if (w.busy) continue; + if (now - w.lastActivity > WORKER_IDLE_TIMEOUT_MS) { + killWorker(senderId); + } + } +} + +// --------------------------------------------------------------------------- +// Attachment download +// --------------------------------------------------------------------------- + +/** Download as raw buffer (for text extraction / base64 image encoding). */ +async function downloadAttachmentRaw(pathId) { + try { + const res = await fetch(`${config.zulip.site}/api/v1/user_uploads/${pathId}`, { + headers: { Authorization: authHeader() }, + signal: AbortSignal.timeout(30000), + }); + if (!res.ok) return null; + const buf = Buffer.from(await res.arrayBuffer()); + return { contentType: res.headers.get("content-type") || "application/octet-stream", buf, size: buf.length }; + } catch (e) { + console.error(`[zulip-ext] Download failed: ${e.message}`); + return null; + } +} + +/** Download and base64-encode (for pi image API). */ +async function downloadAttachment(pathId) { + const raw = await downloadAttachmentRaw(pathId); + if (!raw) return null; + return { contentType: raw.contentType, data: raw.buf.toString("base64"), size: raw.size }; +} + +// --------------------------------------------------------------------------- +// Attachment classification & text extraction +// --------------------------------------------------------------------------- + +const IMAGE_EXTS = new Set(["png", "jpg", "jpeg", "gif", "webp", "bmp", "svg"]); +const PDF_EXTS = new Set(["pdf"]); +const TEXT_EXTS = new Set([ + "txt", "md", "py", "js", "ts", "jsx", "tsx", "json", "yaml", "yml", + "log", "csv", "sh", "bash", "env", "cfg", "conf", "ini", "toml", + "xml", "html", "htm", "css", "scss", "sql", "rs", "go", "java", + "c", "cpp", "cc", "h", "hpp", "rb", "php", "swift", "kt", "scala", + "r", "lua", "pl", "ps1", "bat", "makefile", "dockerfile", "gitignore", + "editorconfig", "diff", "patch", "tex", "rst", "org", "nix", +]); + +const MAX_TEXT_ATTACHMENT_CHARS = 50_000; + +function classifyAttachment(att) { + const name = att.name || ""; + const ext = name.split(".").pop()?.toLowerCase() || ""; + if (IMAGE_EXTS.has(ext)) return "image"; + if (PDF_EXTS.has(ext)) return "pdf"; + if (TEXT_EXTS.has(ext)) return "text"; + // MIME-based fallback + return "binary"; +} + +function formatSize(bytes) { + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + return `${(bytes / (1024 * 1024)).toFixed(1)} MB`; +} + +function tryDecodeText(buf) { + try { return buf.toString("utf-8"); } catch { return null; } +} + +/** Try to extract text from a PDF using pdftotext (poppler-utils). */ +function extractPdfText(buf) { + try { + const tmpIn = `/tmp/zulip-pdf-${Date.now()}.pdf`; + fs.writeFileSync(tmpIn, buf); + const text = execSync(`pdftotext -layout -nopgbrk "${tmpIn}" -`, { + timeout: 15000, + encoding: "utf-8", + maxBuffer: 5 * 1024 * 1024, + }); + fs.unlinkSync(tmpIn); + return text.trim() || null; + } catch (e) { + console.error(`[zulip-ext] PDF extraction failed: ${e.message}`); + return null; + } +} + +// --------------------------------------------------------------------------- +// Message routing +// --------------------------------------------------------------------------- + +/** + * Send a message to a worker and track the pending reply. + */ +async function routeToWorker(senderId, senderName, replyInfo, content, attachments) { + let worker = workers.get(senderId); + if (!worker) { + worker = spawnWorker(senderId); + workers.set(senderId, worker); + } + + // Send placeholder to Zulip + const placeholders = [":robot: _Processing…_", ":hourglass_flowing_sand: _Thinking…_", ":brain: _Generating…_"]; + const placeholder = placeholders[Date.now() % placeholders.length]; + const target = replyInfo.type === "private" + ? { type: "private", to: senderId } + : { type: "stream", to: replyInfo.streamId ?? replyInfo.streamName, subject: replyInfo.topic }; + + let zulipMsgId = null; + if (zulipQueue) { + try { + zulipMsgId = await zulipQueue.sendMessage({ ...target, content: placeholder }); + } catch (e) { + console.warn(`[zulip-ext] Placeholder send failed: ${e.message}`); + } + } + + worker.pendingReplies.push({ + zulipMsgId, + senderId, + senderName, + replyInfo, + createdAt: Date.now(), + }); + + // Build prompt text + const label = replyInfo.type === "stream" + ? `@all-bots in #${replyInfo.streamName}` + : `DM from @${senderName}`; + const promptText = `[Zulip ${label}]: ${content}`; + + // Handle attachments — download, classify, extract content + let images = []; + let attachmentTexts = []; + if (attachments?.length) { + for (const att of attachments) { + const category = classifyAttachment(att); + const raw = await downloadAttachmentRaw(att.path_id); + if (!raw) { + attachmentTexts.push(`[Attachment: ${att.name} — download failed]`); + continue; + } + + switch (category) { + case "image": { + const b64 = raw.buf.toString("base64"); + images.push({ type: "image", data: b64, mimeType: raw.contentType }); + break; + } + case "text": { + const text = tryDecodeText(raw.buf); + if (text) { + const truncated = text.length > MAX_TEXT_ATTACHMENT_CHARS + ? text.slice(0, MAX_TEXT_ATTACHMENT_CHARS) + `\n\n[...truncated at ${MAX_TEXT_ATTACHMENT_CHARS.toLocaleString()} chars, ${(text.length - MAX_TEXT_ATTACHMENT_CHARS).toLocaleString()} chars removed]` + : text; + attachmentTexts.push( + `[Attachment: ${att.name} (${formatSize(raw.size)})]\n\`\`\`${truncated}\n\`\`\`` + ); + } else { + attachmentTexts.push(`[Attachment: ${att.name} (${formatSize(raw.size)}, ${raw.contentType}) — binary content, could not decode as text]`); + } + break; + } + case "pdf": { + const pdfText = extractPdfText(raw.buf); + if (pdfText) { + const truncated = pdfText.length > MAX_TEXT_ATTACHMENT_CHARS + ? pdfText.slice(0, MAX_TEXT_ATTACHMENT_CHARS) + `\n\n[...truncated at ${MAX_TEXT_ATTACHMENT_CHARS.toLocaleString()} chars]` + : pdfText; + attachmentTexts.push( + `[Attachment: ${att.name} (PDF, ${formatSize(raw.size)})]\n\`\`\`${truncated}\n\`\`\`` + ); + } else { + attachmentTexts.push(`[Attachment: ${att.name} (PDF, ${formatSize(raw.size)}) — text extraction failed (pdftotext unavailable or unreadable PDF)]`); + } + break; + } + default: { + attachmentTexts.push(`[Attachment: ${att.name} (${formatSize(raw.size)}, ${raw.contentType}) — binary file, content not extracted]`); + break; + } + } + } + } + + // Merge attachment text into prompt + const fullText = attachmentTexts.length > 0 + ? promptText + "\n\n---\n" + attachmentTexts.join("\n\n") + : promptText; + + // Send to worker + if (worker.busy) { + sendRpc(worker, { type: "steer", message: fullText }); + console.log(`[zulip-ext] Steered to worker ${senderId} for ${senderName}${attachmentTexts.length ? ' with ' + attachmentTexts.length + ' attachment(s)' : ''}`); + } else { + const cmd = { type: "prompt", message: fullText }; + if (images.length > 0) cmd.images = images; + sendRpc(worker, cmd); + const attInfo = []; + if (images.length > 0) attInfo.push(`${images.length} image(s)`); + if (attachmentTexts.length > 0) attInfo.push(`${attachmentTexts.length} file(s)`); + console.log(`[zulip-ext] Routed to worker ${senderId} for ${senderName}${attInfo.length ? ' with ' + attInfo.join(' + ') : ''}`); + } + + worker.busy = true; + worker.lastMessageContent = content; + worker.lastReplyInfo = replyInfo; + worker.lastActivity = Date.now(); + + // Send typing indicator + if (zulipQueue) { + zulipQueue.sendTyping([senderId], "start").catch(() => {}); + } +} + +// --------------------------------------------------------------------------- +// Zulip command handlers +// --------------------------------------------------------------------------- +const ZULIP_COMMANDS = ["status", "retry", "continue", "new", "help", "stop", "model", "compact", "yolo", "verbose", "approve", "deny"]; + +function parseCommand(content) { + const m = content.match(/^\/(\w+)(?:\s+(.+))?$/); + return m ? { cmd: m[1].toLowerCase(), args: m[2] || "" } : null; +} + +/** + * Handle a Zulip command. Returns true if handled, false to pass through as normal message. + */ +async function handleZulipCommand(cmd, args, senderId, senderName, replyInfo) { + const target = replyInfo.type === "private" + ? { type: "private", to: senderId } + : { type: "stream", to: replyInfo.streamId ?? replyInfo.streamName, subject: replyInfo.topic }; + + const send = (content) => { + if (zulipQueue) zulipQueue.sendMessage({ ...target, content }).catch(() => {}); + }; + + switch (cmd) { + // --- Global commands (no worker needed) --- + case "help": { + const cmds = { + status: "Show system status — PM2 processes, containers, uptime", + retry: "Retry the last response", + continue: "Continue the last response if cut off", + new: "Start fresh — clears conversation context", + help: "Show this help message", + stop: "Stop the current response", + model: "Show or change LLM model", + compact: "Compact conversation context", + yolo: "Acknowledge YOLO mode (no confirmations)", + verbose: "Toggle verbose output", + approve: "Approve pending actions (Hermes compat)", + deny: "Deny pending actions (Hermes compat)", + }; + if (args && cmds[args]) { + send(`**/${args}** — ${cmds[args]}\n\nUsage: \`/${args}\``); + } else { + let helpText = "**Abiba Zulip Commands**\n\n"; + for (const [name, desc] of Object.entries(cmds)) { + helpText += `• \`/${name}\` — ${desc}\n`; + } + helpText += `\nSend any message to chat.`; + send(helpText); + } + return true; + } + + case "status": { + let resp = "**Abiba System Status**\n\n**PM2 Processes:**\n"; + try { + const pm2 = execSync("pm2 status --no-color 2>/dev/null", { timeout: 10000, encoding: "utf-8" }); + for (const line of pm2.split("\n")) { + if (line.includes("abiba-")) { + const cells = line.split("│").filter(c => c.trim()); + if (cells.length >= 6) { + const clean = (s) => s.replace(/\x1b\[[0-9;]*m/g, ""); + resp += `• ${clean(cells[1])}: ${clean(cells[5])} (uptime: ${clean(cells[3])})\n`; + } + } + } + } catch (e) { + resp += `_Error: ${e.message}_\n`; + } + resp += `\n**Worker Sessions:** ${workers.size} active\n`; + for (const [sid, w] of workers) { + resp += `• sender_${sid}: ${w.busy ? "busy" : "idle"}, replies pending: ${w.pendingReplies.length}\n`; + } + resp += `\n**Zulip:** connected=${!!zulipQueue} | processed=${msgCount} | echo prevented bots=${BOT_EMAILS.size}`; + send(resp); + return true; + } + + // --- Per-sender commands (need worker) --- + case "retry": { + const worker = workers.get(senderId); + if (!worker?.lastMessageContent) { + send(":warning: No previous message to retry."); + return true; + } + send(":repeat: Retrying last message…"); + routeToWorker(senderId, senderName, worker.lastReplyInfo, worker.lastMessageContent); + return true; + } + + case "continue": { + send(":arrow_forward: Continuing…"); + routeToWorker(senderId, senderName, replyInfo, "Continue your response from where you left off."); + return true; + } + + case "new": { + const worker = workers.get(senderId); + if (worker) { + worker.lastMessageContent = null; + sendRpc(worker, { type: "new_session" }); + } + const topicMsg = args ? `Start fresh, focusing on: ${args}` : "Start fresh."; + send(`:broom: ${topicMsg}\n\n_Previous context cleared._`); + routeToWorker(senderId, senderName, replyInfo, `[System: Session reset. ${topicMsg} Treat this as a fresh conversation.]`); + return true; + } + + case "stop": { + const worker = workers.get(senderId); + if (worker?.busy) { + sendRpc(worker, { type: "abort" }); + send("⏹️ **Stop signal sent.** Current processing interrupted."); + } else { + send(":information_source: No active processing to stop."); + } + return true; + } + + case "model": { + const worker = workers.get(senderId); + if (!worker) { + send("`/model` requires an active conversation. Send a message first."); + return true; + } + sendRpc(worker, { type: "get_available_models" }); + // We'll need to async-read the response. For now, just acknowledge. + // In a follow-up, the response can be captured and sent to Zulip. + send("🔍 Checking available models…"); + return true; + } + + case "compact": { + const worker = workers.get(senderId); + if (worker) sendRpc(worker, { type: "compact" }); + send("🗜️ **Compaction triggered.** Context will be summarized."); + return true; + } + + case "yolo": + send("🤘 YOLO mode acknowledged. All tools execute immediately."); + return true; + + case "verbose": + send("📝 For detailed output, include 'be thorough' or 'show your work' in your message."); + return true; + + case "approve": + case "deny": + send("ℹ️ Pi doesn't have pending approvals — commands execute immediately."); + return true; + + default: + send(`:question: Unknown command \`/${cmd}\`. Try \`/help\`.`); + return true; + } +} + +// --------------------------------------------------------------------------- +// Polling loop +// --------------------------------------------------------------------------- +let zulipQueue = null; +let pollTimer = null; +let connected = false; +let retryCount = 0; +let retryDelay = 5000; +let msgCount = 0; +let skippedCount = 0; +let lastHeartbeatTime = 0; +let lastError = null; + +async function startPolling() { + console.log(`[zulip-ext] Connecting to ${config.zulip.site} as ${config.zulip.email}…`); + try { + zulipQueue = await createZulipQueue(); + connected = true; + retryCount = 0; + retryDelay = 5000; + lastError = null; + await loadBotEmails(zulipQueue); + console.log(`[zulip-ext] Connected, queue=${zulipQueue.queueId}`); + + pollTimer = setInterval(async () => { + if (!zulipQueue) return; + try { + const events = await zulipQueue.poll(); + if (events.length > 0) { + for (const ev of events) { + await processEvent(ev); + } + } + + // Heartbeat + const now = Date.now(); + if (now - lastHeartbeatTime > HEARTBEAT_INTERVAL_MS) { + lastHeartbeatTime = now; + const workerInfo = Array.from(workers.values()).map(w => `${w.senderId}:${w.busy ? "busy" : "idle"}:${w.pendingReplies.length}`).join(","); + console.log(`[zulip-ext] Heartbeat — msgs=${msgCount} skipped=${skippedCount} workers=[${workerInfo}] queue=${zulipQueue.queueId}`); + } + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + const isQueueExpired = msg.includes("BAD_EVENT_QUEUE_ID") || msg.includes("deregistered"); + if (isQueueExpired) { + console.log(`[zulip-ext] Queue expired, reconnecting… (${msg.slice(0, 80)})`); + connected = false; + clearInterval(pollTimer); + pollTimer = null; + setTimeout(startPolling, retryDelay); + } else { + lastError = msg; + console.error(`[zulip-ext] Poll error: ${msg}`); + } + } + }, POLL_INTERVAL_MS); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + lastError = msg; + console.error(`[zulip-ext] Connection failed: ${msg}`); + + // Exponential backoff (capped at ~5 min) + retryDelay = Math.min(5000 * Math.pow(2, retryCount), 300000); + console.log(`[zulip-ext] Reconnecting in ${Math.round(retryDelay / 1000)}s (attempt ${retryCount + 1})…`); + retryCount++; + setTimeout(startPolling, retryDelay); + } +} + +async function processEvent(event) { + if (event.type !== "message") return; + const msg = event.message; + if (!msg) return; + + // Ignore own messages + if (msg.sender_email === config.zulip.email) return; + + // Echo prevention + if (isBotSender(msg.sender_email)) { + skippedCount++; + if (skippedCount % 10 === 1) { + console.log(`[zulip-ext] Echo skip: ${msg.sender_email} (total=${skippedCount})`); + } + return; + } + + const senderName = msg.sender_full_name || msg.sender_email; + const senderId = msg.sender_id; // keep as number for correct API format + + // Private messages (DM-first, ADR-001) + if (msg.type === "private") { + msgCount++; + console.log(`[zulip-ext] DM from ${senderName} (id=${senderId}): ${(msg.content || "").slice(0, 80)}`); + + // Check for commands + const parsed = parseCommand(msg.content); + if (parsed && ZULIP_COMMANDS.includes(parsed.cmd)) { + await handleZulipCommand(parsed.cmd, parsed.args, senderId, senderName, { + type: "private", senderId, senderName, + }); + return; + } + + routeToWorker(senderId, senderName, { type: "private", senderId, senderName }, msg.content, msg.attachments); + return; + } + + // Stream messages: @mention or @all-bots (ADR-005, ADR-006) + const mentionedUserIds = (msg.mentioned_users || []).map(u => u.user_id); + const isDirectMention = zulipQueue?.botUserId && mentionedUserIds.includes(zulipQueue.botUserId); + const isAllBots = isAllBotsMention(msg.content || ""); + + if (msg.type === "stream" && (isDirectMention || isAllBots)) { + msgCount++; + const streamName = typeof msg.display_recipient === "string" ? msg.display_recipient : "unknown"; + const topic = msg.subject || "general"; + console.log(`[zulip-ext] ${isDirectMention ? "@mention" : "@all-bots"} in #${streamName} > ${topic}`); + + const replyInfo = { type: "stream", streamName, topic, streamId: msg.stream_id, senderId, senderName }; + + const parsed = parseCommand(msg.content); + if (parsed && ZULIP_COMMANDS.includes(parsed.cmd)) { + await handleZulipCommand(parsed.cmd, parsed.args, senderId, senderName, replyInfo); + return; + } + + routeToWorker(senderId, senderName, replyInfo, msg.content, msg.attachments); + } +} + +// --------------------------------------------------------------------------- +// Health endpoint +// --------------------------------------------------------------------------- +let healthServer = null; + +function startHealthServer(port) { + const server = http.createServer((req, res) => { + if (req.url === "/health" || req.url === "/") { + const workerList = []; + for (const [sid, w] of workers) { + workerList.push({ + sender_id: sid, + busy: w.busy, + pending_replies: w.pendingReplies.length, + last_activity: Math.floor((Date.now() - w.lastActivity) / 1000), + }); + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + status: connected && zulipQueue ? "ok" : "down", + platform: "pi", + agent: config.agent.name, + zulip: { + connected, + site: config.zulip.site, + email: config.zulip.email, + queue_id: zulipQueue?.queueId ?? null, + bot_user_id: zulipQueue?.botUserId ?? null, + messages_processed: msgCount, + skipped: skippedCount, + last_error: lastError, + }, + workers: workerList, + worker_count: workers.size, + })); + } else { + res.writeHead(404); + res.end("Not found"); + } + }); + server.on("error", (err) => { + if (err.code === "EADDRINUSE") { + console.warn(`[zulip-ext] Port :${port} in use — skipping health`); + } + }); + server.listen(port, "127.0.0.1", () => { + console.log(`[zulip-ext] Health endpoint on :${port}`); + }); + return server; +} + +// --------------------------------------------------------------------------- +// Extension lifecycle +// --------------------------------------------------------------------------- +export default function (pi) { + // Guard: only activate in the designated PM2 router process + if (process.env.ZULIP_ROLE !== "router") { + // Worker processes and TUI sessions: NO-OP + return; + } + + pi.on("session_start", async () => { + console.log(`[zulip-ext] Router starting for ${config.agent.name}`); + + // Load persisted worker session map + if (!fs.existsSync(SESSION_DIR)) fs.mkdirSync(SESSION_DIR, { recursive: true }); + workerSessions = loadWorkerSessions(); + + // Start health server + if (!healthServer) { + healthServer = startHealthServer(config.health_port); + } + + // Start Zulip poller + startPolling(); + + // Start idle worker cleanup (every 5 min) + idleCleanupTimer = setInterval(cleanupIdleWorkers, 5 * 60 * 1000); + + console.log(`[zulip-ext] Router ready. Known sessions: ${Object.keys(workerSessions).length}`); + }); + + pi.on("session_shutdown", async () => { + console.log(`[zulip-ext] Router shutting down…`); + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = null; + } + if (idleCleanupTimer) { + clearInterval(idleCleanupTimer); + idleCleanupTimer = null; + } + + // Kill all workers gracefully + for (const [sid, w] of workers) { + try { w.process.kill("SIGTERM"); } catch {} + } + workers.clear(); + + // Delete Zulip event queue + if (zulipQueue?.queueId) { + try { + await fetch(`${config.zulip.site}/api/v1/queues/${zulipQueue.queueId}`, { + method: "DELETE", + headers: { Authorization: authHeader() }, + }); + console.log(`[zulip-ext] Queue ${zulipQueue.queueId} deregistered`); + } catch (e) { + console.warn(`[zulip-ext] Queue deregister failed: ${e.message}`); + } + } + + if (healthServer) { + healthServer.close(); + healthServer = null; + } + connected = false; + zulipQueue = null; + + saveWorkerSessions(); + }); + + // Register diagnostics commands + pi.registerCommand("zulip-status", { + description: "Show Zulip extension status and worker pool", + handler: async (_args, ctx) => { + const lines = [ + "=== Zulip Extension v2 (router-worker) ===", + `Agent: ${config.agent.name}`, + `Server: ${config.zulip.site}`, + `Connected: ${connected ? "✅" : "❌"} Queue: ${zulipQueue?.queueId ?? "N/A"}`, + `Messages: ${msgCount} processed, ${skippedCount} echo-skipped`, + `Workers: ${workers.size} active`, + `Retries: ${retryCount} Last error: ${lastError ?? "none"}`, + "", + ]; + for (const [sid, w] of workers) { + lines.push(`• sender_${sid}: ${w.busy ? "busy" : "idle"}, ${w.pendingReplies.length} pending, session=${w.sessionFile ?? "new"}`); + } + ctx.ui.notify(lines.join("\n"), "info"); + }, + }); + + pi.registerCommand("zulip-send-test", { + description: "Send test DM: /zulip-send-test ", + handler: async (args, ctx) => { + const parts = args.trim().match(/^([^\s]+)\s+(.+)$/); + const to = parts?.[1] ?? config.agent.owner_email; + const msg = parts?.[2] ?? "Test from pi Zulip extension v2"; + if (!zulipQueue) { + ctx.ui.notify("Not connected. Check /zulip-status.", "error"); + return; + } + try { + await zulipQueue.sendMessage({ type: "private", to, content: msg }); + ctx.ui.notify(`Test sent to ${to}`, "info"); + } catch (err) { + ctx.ui.notify(`Failed: ${err.message}`, "error"); + } + }, + }); + + pi.registerCommand("zulip-test", { + description: "Run comprehensive Zulip self-test", + handler: async (_args, ctx) => { + const results = []; + const passed = (n, d) => results.push({ name: n, ok: true, detail: d }); + const failed = (n, d) => results.push({ name: n, ok: false, detail: d }); + + passed("queue_connected", `Queue: ${zulipQueue?.queueId ?? "N/A"}`); + zulipQueue?.botUserId ? passed("bot_identity", `Bot ID: ${zulipQueue.botUserId}`) : failed("bot_identity", "Bot ID not resolved"); + zulipQueue && zulipQueue.allBotsUserId > 1 ? passed("all_bots", `ID: ${zulipQueue.allBotsUserId}`) : failed("all_bots", `ID: ${zulipQueue?.allBotsUserId ?? 1}`); + BOT_EMAILS.size > 1 ? passed("echo_prevention", `${BOT_EMAILS.size} bot emails`) : failed("echo_prevention", "Only self excluded"); + workers.size >= 0 ? passed("worker_pool", `${workers.size} active workers`) : failed("worker_pool", "Pool broken"); + + // Loopback send + if (zulipQueue) { + try { + const id = await zulipQueue.sendMessage({ type: "private", to: config.agent.owner_email, content: `🔬 Zulip self-test at ${new Date().toISOString()}` }); + passed("api_send", `Loopback DM sent (id=${id})`); + } catch (e) { + failed("api_send", e.message.slice(0, 80)); + } + } + + const ok = results.filter(r => r.ok).length; + const total = results.length; + const verdict = ok === total ? "HEALTHY" : ok >= total - 1 ? "DEGRADED" : "CRITICAL"; + let report = `=== Zulip Self-Test: ${verdict} ===\n${ok}/${total} checks passed\n\n`; + for (const r of results) { + report += `${r.ok ? "✅" : "❌"} ${r.name}: ${r.detail}\n`; + } + ctx.ui.notify(report, verdict === "HEALTHY" ? "info" : "error"); + }, + }); +} diff --git a/pi-zulip-extension/src/mcp/index.ts b/pi-zulip-extension/src/mcp/index.ts new file mode 100644 index 0000000..3cc8524 --- /dev/null +++ b/pi-zulip-extension/src/mcp/index.ts @@ -0,0 +1,460 @@ +/** + * MCP Bridge Extension + * + * Connects to MCP servers and exposes their tools as pi custom tools. + * Supports stdio and streaming-HTTP transports. + * + * Configuration is read from ~/.pi/agent/extensions/mcp/mcp-servers.json. + * See the README in this directory for the format. + */ + +import type { ExtensionAPI, ExtensionContext } from "@mariozechner/pi-coding-agent"; +import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { + CallToolResultSchema, + ListToolsResultSchema, +} from "@modelcontextprotocol/sdk/types.js"; +import { Type } from "typebox"; +import fs from "node:fs"; +import path from "node:path"; + +// --------------------------------------------------------------------------- +// Configuration +// --------------------------------------------------------------------------- + +interface McpStdioServer { + command: string; + args?: string[]; + env?: Record; + /** Whitelist: only register these tool names. Takes priority over excludeTools. */ + tools?: string[]; + /** Blacklist: exclude these tool names. Ignored if `tools` is set. */ + excludeTools?: string[]; +} + +interface McpHttpServer { + name: string; + description?: string; + type: "streaming-http"; + url: string; + timeout?: number; + /** Whitelist: only register these tool names. Takes priority over excludeTools. */ + tools?: string[]; + /** Blacklist: exclude these tool names. Ignored if `tools` is set. */ + excludeTools?: string[]; +} + +type McpServerConfig = McpStdioServer | McpHttpServer; + +interface McpConfig { + mcpServers: Record; +} + +function loadConfig(): McpConfig { + const configPath = path.join( + path.dirname(import.meta.url.replace("file://", "")), + "mcp-servers.json", + ); + if (!fs.existsSync(configPath)) { + return { mcpServers: {} }; + } + return JSON.parse(fs.readFileSync(configPath, "utf-8")); +} + +// --------------------------------------------------------------------------- +// MCP client registry +// --------------------------------------------------------------------------- + +interface ServerClient { + client: Client; + transport: StdioClientTransport | StreamableHTTPClientTransport; + name: string; + tools: Array<{ name: string; description: string; inputSchema: object }>; +} + +const clients = new Map(); + +async function connectToServer( + name: string, + config: McpServerConfig, + ctx: ExtensionContext, +): Promise { + let client: Client; + let transport: StdioClientTransport | StreamableHTTPClientTransport; + + if ("command" in config) { + // stdio transport + transport = new StdioClientTransport({ + command: config.command, + args: config.args ?? [], + env: { + ...process.env, + ...config.env, + }, + }); + client = new Client( + { name: "pi-mcp", version: "1.0.0" }, + { capabilities: {} }, + ); + await client.connect(transport); + } else { + // streaming-HTTP transport + transport = new StreamableHTTPClientTransport( + new URL(config.url), + undefined, + { + requestTimeout: config.timeout ?? 120_000, + }, + ); + client = new Client( + { name: "pi-mcp", version: "1.0.0" }, + { capabilities: {} }, + ); + await client.connect(transport); + } + + // Discover tools + const toolsRes = await client.request( + { method: "tools/list" }, + ListToolsResultSchema, + ); + + const tools = + toolsRes.tools?.map((t) => ({ + name: t.name, + description: t.description ?? "", + inputSchema: t.inputSchema ?? {}, + })) ?? []; + + return { client, transport, name, tools }; +} + +async function disconnectClient(server: ServerClient) { + try { + await server.transport.close(); + } catch { + // ignore close errors + } +} + +// --------------------------------------------------------------------------- +// Schema builder: convert MCP inputSchema to TypeBox +// --------------------------------------------------------------------------- + +function buildParameters(inputSchema: object) { + if (!inputSchema || typeof inputSchema !== "object") { + return Type.Object({}); + } + + const schema = inputSchema as { + type?: string; + properties?: Record; + required?: string[]; + }; + + const properties: Record = {}; + + if (schema.properties) { + for (const [key, prop] of Object.entries(schema.properties)) { + const p = prop as { type?: string; description?: string }; + let field: any; + switch (p.type) { + case "string": + field = Type.Optional(Type.String({ description: p.description })); + break; + case "number": + field = Type.Optional(Type.Number()); + break; + case "integer": + field = Type.Optional(Type.Number()); + break; + case "boolean": + field = Type.Optional(Type.Boolean()); + break; + case "array": + field = Type.Optional( + Type.Array(Type.String(), { + description: p.description, + }), + ); + break; + case "object": + field = Type.Optional(Type.Record(Type.String(), Type.Unknown())); + break; + default: + field = Type.Optional(Type.String()); + } + properties[key] = field; + } + } + + // Check required fields + if (schema.required) { + for (const req of schema.required) { + if (properties[req]) { + // Make required by redefining + const def = properties[req]; + if (def && "origin" in def) { + // It's a TypeBox optional - we'd need to rebuild as non-optional + // For simplicity, keep all optional since MCP tools often have flexible params + } + } + } + } + + return Type.Object(properties); +} + +// --------------------------------------------------------------------------- +// Tool call forwarder +// --------------------------------------------------------------------------- + +const DEFAULT_MAX_RESULT_CHARS = 15_000; + +/** + * Rebuild relay query text from structuredContent when the bridge has + * truncated the display text field (ends with Unicode ellipsis character). + */ +function rebuildRelayTextFromStructuredContent( + textContent: string, + structuredContent: Record | undefined, +): string | null { + if (!structuredContent) return null; + + // Detect bridge-side truncation: the display text ends with … + if (!textContent.endsWith("…")) return null; + + const nodes = structuredContent.nodes as Array> | undefined; + if (!nodes || nodes.length === 0) return null; + + const rebuilt: string[] = []; + rebuilt.push(`📬 **Relay Query**: ${structuredContent.count ?? nodes.length} result(s).\n`); + + for (const node of nodes) { + const id = node.id; + const title = node.title ?? "(no title)"; + const from = (node.metadata as Record)?.from ?? "?"; + const to = (node.metadata as Record)?.to ?? "?"; + const source = node.source ?? ""; + const description = node.description ?? ""; + + rebuilt.push(`**#${id}** — ${title}`); + rebuilt.push(` From: ${from} → To: ${to}`); + if (source) { + rebuilt.push(` ${source.replace(/\n/g, "\n ")}`); + } + if (description) { + rebuilt.push(` ${description}`); + } + rebuilt.push(""); + } + + return rebuilt.join("\n"); +} + +async function forwardToolCall( + server: ServerClient, + toolName: string, + args: Record, + signal?: AbortSignal, + maxResultChars: number = DEFAULT_MAX_RESULT_CHARS, +): Promise<{ + content: Array<{ type: string; text: string }>; + isError?: boolean; +}> { + const result = await server.client.request( + { + method: "tools/call", + params: { name: toolName, arguments: args }, + }, + CallToolResultSchema, + signal, + ); + + // Check for bridge-side truncation and rebuild from structuredContent + const structuredContent = (result as Record).structuredContent as Record | undefined; + + const content: Array<{ type: string; text: string }> = []; + let totalChars = 0; + let truncated = false; + + if (result.content) { + for (const item of result.content) { + if (truncated) break; + if (item.type === "text") { + let itemText = item.text; + + // If bridge-side truncation detected, rebuild from structuredContent + const rebuilt = rebuildRelayTextFromStructuredContent(itemText, structuredContent); + if (rebuilt) { + itemText = rebuilt; + } + + totalChars += itemText.length; + if (totalChars > maxResultChars) { + const keep = maxResultChars - (totalChars - itemText.length); + const trimmed = itemText.slice(0, Math.max(0, keep)); + content.push({ + type: "text", + text: trimmed + `\n\n[TRUNCATED: ${(itemText.length - keep).toLocaleString()} chars removed. Limit: ${maxResultChars.toLocaleString()} chars. Use specific queries or pagination.]`, + }); + truncated = true; + } else { + content.push({ type: "text", text: itemText }); + } + } else if (item.type === "image") { + content.push({ + type: "text", + text: `[MCP image: ${item.mimeType}]`, + }); + } else if (item.type === "resource") { + const textParts: string[] = []; + if ("resource" in item) { + const res = item.resource; + if ("text" in res) { + textParts.push(res.text); + } else if ("blob" in res) { + textParts.push(`[binary resource: ${res.mimeType}]`); + } + } + content.push({ type: "text", text: textParts.join("\n") }); + } else { + content.push({ + type: "text", + text: `[MCP ${item.type}: ${JSON.stringify(item).slice(0, 500)}]`, + }); + } + } + } + + return { + content, + isError: result.isError ?? false, + }; +} + +// --------------------------------------------------------------------------- +// Extension +// --------------------------------------------------------------------------- + +export default async function (pi: ExtensionAPI) { + const config = loadConfig(); + const registeredNames = new Set(); + + // Connect to all servers + for (const [name, serverConfig] of Object.entries(config.mcpServers)) { + try { + const client = await connectToServer(name, serverConfig, pi); + + if (client.tools.length === 0) { + pi.on("session_start", (_event, ctx) => { + ctx.ui.notify( + `MCP server "${name}" connected but has no tools.`, + "info", + ); + }); + clients.set(name, client); + continue; + } + + // Filter tools based on server config + let registeredTools = client.tools; + if (serverConfig.tools) { + registeredTools = client.tools.filter((t) => + serverConfig.tools!.includes(t.name), + ); + } else if (serverConfig.excludeTools) { + registeredTools = client.tools.filter( + (t) => !serverConfig.excludeTools!.includes(t.name), + ); + } + + pi.on("session_start", (_event, ctx) => { + ctx.ui.notify( + `MCP: "${name}" connected — ${registeredTools.length}/${client.tools.length} tool(s) registered.`, + "info", + ); + }); + + // Register each MCP tool as a pi tool + for (const tool of registeredTools) { + const fullName = `${name}-${tool.name}`; + if (registeredNames.has(fullName)) continue; + registeredNames.add(fullName); + + const parameters = buildParameters(tool.inputSchema); + + pi.registerTool({ + name: fullName, + label: tool.name, + description: `[MCP:${name}] ${tool.description}`, + promptSnippet: `[MCP:${name}] ${tool.description}`, + promptGuidelines: [ + `Use ${fullName} to: ${tool.description}.`, + ], + parameters, + async execute( + _toolCallId, + params: Record, + signal, + _onUpdate, + _ctx, + ) { + const server = clients.get(name); + if (!server) { + return { + content: [{ type: "text", text: `Server "${name}" not connected.` }], + isError: true, + }; + } + + const maxChars = serverConfig.maxResultChars ?? DEFAULT_MAX_RESULT_CHARS; + return await forwardToolCall(server, tool.name, params, signal, maxChars); + }, + }); + } + + clients.set(name, client); + } catch (err) { + pi.on("session_start", (_event, ctx) => { + ctx.ui.notify( + `MCP: Failed to connect to "${name}": ${err instanceof Error ? err.message : String(err)}`, + "error", + ); + }); + } + } + + // Register a /mcp-status command + pi.registerCommand("mcp-status", { + description: "Show status of all MCP servers", + handler: async (_args, ctx) => { + const lines: string[] = ["=== MCP Server Status ===", ""]; + for (const [name, client] of clients) { + lines.push(` ${name}: connected (${client.tools.length} tools)`); + for (const tool of client.tools) { + lines.push(` - ${tool.name}: ${tool.description.slice(0, 80)}`); + } + } + const unconnected = Object.keys(config.mcpServers).filter( + (n) => !clients.has(n), + ); + for (const name of unconnected) { + lines.push(` ${name}: NOT CONNECTED`); + } + ctx.ui.notify(lines.join("\n"), "info"); + }, + }); + + // Cleanup on shutdown + pi.on("session_shutdown", async (event, _ctx) => { + if (event.reason === "quit") { + for (const [name, client] of clients) { + await disconnectClient(client); + delete clients.get(name); + } + } + }); +} diff --git a/plugins/platforms/zulip/adapter.py b/plugins/platforms/zulip/adapter.py index d84685b..1e25176 100644 --- a/plugins/platforms/zulip/adapter.py +++ b/plugins/platforms/zulip/adapter.py @@ -100,6 +100,23 @@ def _build_auth_header(email: str, api_key: str) -> str: return f"Basic {token}" +def _strip_html(text: str) -> str: + """Strip HTML tags and decode HTML entities from Zulip message content. + + Zulip delivers message content as rendered HTML (e.g.

/approve

). + The gateway slash command parser expects plain text, so HTML tags + prevent command matching. This helper strips tags and decodes entities. + """ + if not text: + return text + # Strip HTML tags + text = re.sub(r"<[^>]+>", "", text) + # Decode common HTML entities + text = text.replace("&", "&").replace("<", "<").replace(">", ">") + text = text.replace(""", '"').replace("'", "'").replace(" ", " ") + return text.strip() + + def _truncate(text: str, limit: int = MAX_ZULIP_MESSAGE) -> str: """Truncate to Zulip's message limit with notice.""" if len(text) <= limit: @@ -303,6 +320,9 @@ class ZulipAdapter(BasePlatformAdapter): # Gen 2: Try to resolve @all-bots user ID dynamically await self._resolve_all_bots_user_id() + # Gen 5: Ensure bot is subscribed to primary stream + await self._ensure_stream_subscriptions() + self._mark_connected() logger.info( "[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)", @@ -450,6 +470,10 @@ class ZulipAdapter(BasePlatformAdapter): 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: @@ -678,6 +702,64 @@ class ZulipAdapter(BasePlatformAdapter): # Gen 2: Dynamic @all-bots resolution # ------------------------------------------------------------------ + async def _ensure_stream_subscriptions(self) -> None: + """Ensure the bot is subscribed to the primary streams. + + Checks current subscriptions and subscribes to the configured primary + stream (and general-chat) if missing. Without subscription, the bot + receives DMs but no stream events — @mentions in streams are invisible. + + This is a Gen 5 improvement based on lessons from the pi Zulip extension, + where the bot had zero stream subscriptions and couldn't receive + @mentions in stream topics. + """ + try: + # Check current subscriptions + resp, _ = await self._api_call( + "GET", "/api/v1/users/me/subscriptions", + ) + if not resp: + return + + subscribed = resp.get("subscriptions", []) + subscribed_names = [s.get("name", "") for s in subscribed if isinstance(s, dict)] + + # Streams the bot should be subscribed to + required_streams = [self._stream, "general chat"] + missing = [s for s in required_streams if s not in subscribed_names] + + if not missing: + logger.info( + "[%s] Stream subscriptions OK: %s", + self.name, ", ".join(subscribed_names), + ) + return + + # Subscribe to missing streams + import json as _json + payload = {"subscriptions": _json.dumps( + [{"name": s} for s in missing] + )} + sub_resp, _ = await self._api_call( + "POST", "/api/v1/users/me/subscriptions", + data=payload, + ) + if sub_resp and sub_resp.get("result") == "success": + subscribed_str = ", ".join(missing) + logger.info( + "[%s] Subscribed to: %s", self.name, subscribed_str, + ) + self._health_stats["subscriptions_fixed"] = ( + self._health_stats.get("subscriptions_fixed", 0) + 1 + ) + else: + logger.warning( + "[%s] Failed to subscribe to %s", + self.name, ", ".join(missing), + ) + except Exception as e: + logger.warning("[%s] Subscription check failed: %s", self.name, e) + async def _resolve_all_bots_user_id(self) -> None: """Try to resolve the @all-bots user ID from the Zulip server. @@ -720,6 +802,8 @@ class ZulipAdapter(BasePlatformAdapter): sender_name = msg.get("sender_full_name", "Unknown") sender_id = msg.get("sender_id") content = msg.get("content", "") + # Strip Zulip HTML for slash command matching (zulip-approval-fix contract) + content = _strip_html(content) # Echo-loop prevention: skip own messages if sender_email == self._bot_email: @@ -1097,6 +1181,35 @@ class ZulipAdapter(BasePlatformAdapter): "detail": f"@all-bots user_id: {self._all_bots_user_id}", } + # 9. Stream subscriptions (verifies bot is subscribed to at least one stream) + try: + from urllib.parse import urlencode + import json as _json + key_str = f"{self._email}:{self._api_key}" + auth_b64 = base64.b64encode(key_str.encode()).decode() + resp, _ = await self._api_call( + "GET", "/api/v1/users/me/subscriptions", + ) + if resp and "subscriptions" in resp: + stream_count = len(resp["subscriptions"]) + stream_names = [s.get("name", "?") for s in resp["subscriptions"]][:5] + checks["stream_subscriptions"] = { + "status": stream_count > 0, + "detail": f"{stream_count} stream(s): {', '.join(stream_names)}" + if stream_count > 0 + else "No stream subscriptions — bot will not receive stream events", + } + else: + checks["stream_subscriptions"] = { + "status": False, + "detail": "Failed to check subscriptions", + } + except Exception as sub_err: + checks["stream_subscriptions"] = { + "status": False, + "detail": f"Subscription check failed: {sub_err}", + } + # Overall verdict critical = ["connected", "queue_registered", "http_client", "poll_loop"] passed = sum(1 for c in checks.values() if c["status"])