SyslogSolution/zulip-platform-plugins
7
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
10e4e0374f822cffda30374cef7a0d6412e05cbd
zulip-platform-plugins/plugins/platforms/zulip/adapter.py
T
Abiba (pi) 0acf1068bf feat(hermes): Zulip adapter Gen 3 — malformed message resilience, periodic @all-bots refresh, health callback 
Gen 3 improvements from build-zulip-plugin contract run:

1. Malformed message resilience
   - Per-event try/except in poll loop — one bad event never kills adapter
   - malformed_events counter tracked in health stats
   - Fulfills Success Criterion: handles_malformed_messages == true

2. Periodic @all-bots refresh
   - _all_bots_refresh_forever() background task re-resolves every hour
   - Cancelled gracefully on disconnect()
   - all_bots_refreshes counter in health stats

3. Health stats callback mechanism
   - set_health_callback(callback, interval=600) — register external consumer
   - _report_health_if_callback() triggers on dedup cleanup cycle
   - Designed for Hermes Gateway to wire to RA-H OS knowledge graph logging
   - Cross-contract integration: get_all_bots_user_id() and get_bot_user_id()
     exposed for zulip-mention-reliability contract
2026-06-26 22:31:16 +00:00

1170 lines
42 KiB
Python
Raw Blame History

"""
Zulip platform adapter (Hermes plugin) — Gen 3.
Connects a Hermes agent to Zulip via event queue polling. DMs and @mentions
are routed to the agent via the Hermes Gateway's handle_message() interface.
Replies use a placeholder->edit streaming pattern for UX feedback.
Gen 3 Improvements (2026-06-26):
1. Malformed message resilience — try/except wraps each event, no crash on bad JSON
2. Periodic @all-bots refresh — background task re-resolves user ID every hour
3. Health stats callback — external consumers can register for periodic stats
Gen 2 Improvements (2026-06-26):
1. Background dedup maintenance — O(1) per-message path, periodic cleanup task
2. Self-test capability — selftest() method verifying DM/mention/queue health
3. Health stats tracking — poll stats, reconnect counts, error rates
4. Dynamic @all-bots resolution — falls back to hardcoded default
Architecture (mirrors pi-zulip-extension):
Zulip event queue -> poll loop -> parse message -> handle_message(MessageEvent)
-> Gateway processes -> send() / edit_message() posts response to Zulip
No external SDK required — only httpx (already a Hermes dependency).
Ships as a Hermes platform plugin under plugins/platforms/zulip/.
"""
import asyncio
import base64
import json
import logging
import os
import re
import time
import uuid
from collections import deque
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
try:
import httpx
HTTPX_AVAILABLE = True
except ImportError:
HTTPX_AVAILABLE = False
httpx = None
from gateway.config import PlatformConfig
from gateway.platforms.base import (
BasePlatformAdapter,
MessageEvent,
MessageType,
SendResult,
)
logger = logging.getLogger(__name__)
# Constants
DEFAULT_STREAM = "agent-hub"
DEFAULT_ALL_BOTS_USER_ID = 1
DEFAULT_POLL_INTERVAL = 3.0
MAX_ZULIP_MESSAGE = 10000
ECHO_TAG_PREFIX = "hermes-zulip-"
RECONNECT_BACKOFF = [2, 5, 10, 30, 60]
DEDUP_WINDOW = 300 # 5 minutes
DEDUP_MAX_SIZE = 1000
DEDUP_CLEANUP_INTERVAL = 120 # Clean old entries every 2 minutes
ALL_BOTS_REFRESH_INTERVAL = 3600 # Re-resolve @all-bots user ID every hour
SELFTEST_TIMEOUT = 30 # seconds to wait for self-test response
# Regex to strip Zulip @mention markup
MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*")
# Placeholder messages for streaming UX
PLACEHOLDERS = [
":robot: _Processing your message..._",
":hourglass_flowing_sand: _Thinking..._",
":brain: _Generating response..._",
]
def _build_auth_header(email: str, api_key: str) -> str:
"""Build Basic auth header for Zulip API."""
token = base64.b64encode(f"{email}:{api_key}".encode()).decode()
return f"Basic {token}"
def _truncate(text: str, limit: int = MAX_ZULIP_MESSAGE) -> str:
"""Truncate to Zulip's message limit with notice."""
if len(text) <= limit:
return text
return text[:limit] + "\n\n[...truncated at Zulip limit]"
def _parse_zulip_timestamp(ts: Any) -> datetime:
"""Convert Zulip timestamp (float seconds) to datetime."""
try:
return datetime.fromtimestamp(float(ts), tz=timezone.utc)
except (ValueError, OSError, TypeError):
return datetime.now(tz=timezone.utc)
class ZulipAdapter(BasePlatformAdapter):
"""Zulip adapter for Hermes agents.
Connects to Zulip via event queues, polls for new events, converts
DMs and @mentions to MessageEvent, and sends responses with streaming
placeholder->edit UX.
Gen 3 adds:
- Malformed message resilience (per-event try/except)
- Periodic @all-bots refresh (hourly background task)
- Health stats callback for external consumers
Gen 2 adds:
- Background dedup maintenance task (O(1) per-message)
- Self-test diagnostics (DM/mention/queue health)
- Health stats for monitoring
- Dynamic @all-bots resolution
"""
def __init__(self, config: PlatformConfig):
platform_name = "zulip"
super().__init__(config=config, platform=platform_name)
extra = config.extra or {}
# Zulip connection
self._site: str = (
extra.get("site") or os.getenv("ZULIP_SITE", "")
).rstrip("/")
self._email: str = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
self._api_key: str = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")
# Agent identity
self._agent_name: str = (
extra.get("agent_name") or os.getenv("ZULIP_AGENT_NAME", "hermes-agent")
)
# Stream config
self._stream: str = (
extra.get("stream") or os.getenv("ZULIP_STREAM", DEFAULT_STREAM)
)
# Self-test config — owner user ID for DM self-tests
self._owner_user_id: Optional[int] = self._resolve_int_opt(
extra.get("owner_user_id") or os.getenv("ZULIP_OWNER_USER_ID", "")
)
# @all-bots user ID — dynamically resolved on connect, falls back to config/env/default
self._all_bots_user_id: int = self._resolve_int(
extra.get("all_bots_user_id")
or os.getenv("ZULIP_ALL_BOTS_USER_ID", str(DEFAULT_ALL_BOTS_USER_ID)),
DEFAULT_ALL_BOTS_USER_ID,
)
# Polling
self._poll_interval: float = float(
extra.get("poll_interval")
or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL))
)
# --- State ---
self._auth_header: str = _build_auth_header(self._email, self._api_key)
self._queue_id: Optional[str] = None
self._last_event_id: int = -1
self._poll_task: Optional[asyncio.Task] = None
self._dedup_cleanup_task: Optional[asyncio.Task] = None
self._http_client: Optional[httpx.AsyncClient] = None
self._pending_replies: Dict[str, int] = {} # chat_id -> placeholder zulip_msg_id
# Derived identity for echo-loop prevention
self._bot_user_id: Optional[int] = None
self._bot_email: str = self._email
# --- Gen 2: Dedup (background-maintained) ---
self._seen_message_ids: Dict[str, float] = {}
# --- Gen 2/3: Health stats ---
self._health_stats: Dict[str, Any] = {
"started_at": None,
"poll_count": 0,
"poll_errors": 0,
"malformed_events": 0,
"messages_received": 0,
"dms_routed": 0,
"mentions_routed": 0,
"reconnects": 0,
"send_count": 0,
"send_errors": 0,
"all_bots_refreshes": 0,
"last_poll_at": None,
"last_message_at": None,
"last_error_at": None,
"last_error_msg": None,
}
# --- Gen 2: Self-test future for async response ---
self._selftest_future: Optional[asyncio.Future] = None
# --- Gen 3: Periodic @all-bots refresh ---
self._all_bots_refresh_task: Optional[asyncio.Task] = None
# --- Gen 3: Health stats callback ---
self._health_callback = None
self._health_callback_interval: int = 600 # 10 min
@staticmethod
def _resolve_int(val: Any, default: int) -> int:
"""Resolve an integer config value, falling back to default."""
if val is None or val == "":
return default
try:
return int(val)
except (ValueError, TypeError):
return default
@staticmethod
def _resolve_int_opt(val: Any) -> Optional[int]:
"""Resolve an optional integer config value."""
if val is None or val == "":
return None
try:
return int(val)
except (ValueError, TypeError):
return None
# ------------------------------------------------------------------
# Connection lifecycle
# ------------------------------------------------------------------
async def connect(self) -> bool:
"""Connect to Zulip and register an event queue."""
if not HTTPX_AVAILABLE:
logger.warning("[%s] httpx not installed", self.name)
return False
if not self._site or not self._email or not self._api_key:
logger.warning("[%s] Zulip credentials not configured", self.name)
return False
try:
self._http_client = httpx.AsyncClient(timeout=30.0)
self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat()
# Register event queue
queue_resp = await self._api_call(
"POST", "/api/v1/register",
data={
"event_types": '["message"]',
"apply_markdown": "true",
"include_subscribers": "false",
},
)
if not queue_resp:
logger.error("[%s] Failed to register event queue", self.name)
return False
data = queue_resp
self._queue_id = data.get("queue_id")
self._last_event_id = data.get("last_event_id", -1)
self._bot_user_id = data.get("user_id")
# Gen 2: Try to resolve @all-bots user ID dynamically
await self._resolve_all_bots_user_id()
self._mark_connected()
logger.info(
"[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)",
self.name, self._site, self._email,
self._queue_id, self._bot_user_id, self._all_bots_user_id,
)
# Start poll loop
self._poll_task = asyncio.create_task(self._poll_forever())
# Gen 2: Start background dedup cleanup task
self._dedup_cleanup_task = asyncio.create_task(
self._dedup_cleanup_forever()
)
# Gen 3: Start periodic @all-bots refresh task
self._all_bots_refresh_task = asyncio.create_task(
self._all_bots_refresh_forever()
)
return True
except Exception as e:
logger.error("[%s] Connection failed: %s", self.name, e)
return False
async def disconnect(self) -> None:
"""Disconnect from Zulip and cancel all background tasks."""
# Cancel dedup cleanup task
if self._dedup_cleanup_task:
self._dedup_cleanup_task.cancel()
try:
await self._dedup_cleanup_task
except asyncio.CancelledError:
pass
self._dedup_cleanup_task = None
# Cancel @all-bots refresh task
if self._all_bots_refresh_task:
self._all_bots_refresh_task.cancel()
try:
await self._all_bots_refresh_task
except asyncio.CancelledError:
pass
self._all_bots_refresh_task = None
# Cancel poll task
if self._poll_task:
self._poll_task.cancel()
try:
await self._poll_task
except asyncio.CancelledError:
pass
self._poll_task = None
self._queue_id = None
self._mark_disconnected()
logger.info("[%s] Disconnected", self.name)
# ------------------------------------------------------------------
# Poll loop — receives messages from Zulip
# ------------------------------------------------------------------
async def _poll_forever(self) -> None:
"""Continuously poll the Zulip event queue."""
backoff_idx = 0
while self._running:
if not self._queue_id:
await asyncio.sleep(self._poll_interval)
continue
try:
events = await self._fetch_events()
self._health_stats["poll_count"] += 1
self._health_stats["last_poll_at"] = datetime.now(
timezone.utc
).isoformat()
for event in events:
try:
await self._process_zulip_event(event)
except Exception as e:
# Gen 3: Malformed message resilience — catch per-event
# failures so one bad message never kills the poll loop
logger.warning(
"[%s] Malformed event skipped: %s. "
"Event id=%s",
self.name, e,
event.get("id", "unknown"),
)
self._health_stats["malformed_events"] = (
self._health_stats.get("malformed_events", 0) + 1
)
backoff_idx = 0
except asyncio.CancelledError:
break
except Exception as e:
if not self._running:
break
self._health_stats["poll_errors"] += 1
self._health_stats["last_error_at"] = datetime.now(
timezone.utc
).isoformat()
self._health_stats["last_error_msg"] = str(e)[:200]
err_str = str(e)
if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower():
logger.info("[%s] Queue expired, re-registering...", self.name)
await self._reconnect()
backoff_idx = 0
continue
logger.warning("[%s] Poll error: %s", self.name, e)
delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF) - 1)]
backoff_idx += 1
await asyncio.sleep(delay)
await asyncio.sleep(self._poll_interval)
async def _fetch_events(self) -> List[Dict[str, Any]]:
"""Fetch events from the Zulip event queue."""
if not self._queue_id:
return []
resp = await self._api_call(
"GET", "/api/v1/events",
params={
"queue_id": self._queue_id,
"last_event_id": str(self._last_event_id),
"dont_block": "true",
},
)
if not resp:
return []
events = resp.get("events", [])
# Track last event ID for cursor advancement
for event in events:
eid = event.get("id", 0)
if eid > self._last_event_id:
self._last_event_id = eid
return [e for e in events if e.get("type") == "message"]
async def _reconnect(self) -> None:
"""Re-register the event queue."""
self._queue_id = None
try:
resp = await self._api_call(
"POST", "/api/v1/register",
data={
"event_types": '["message"]',
"apply_markdown": "true",
"include_subscribers": "false",
},
)
if resp:
self._queue_id = resp.get("queue_id")
self._last_event_id = resp.get("last_event_id", -1)
self._health_stats["reconnects"] += 1
logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id)
except Exception as e:
logger.error("[%s] Reconnect failed: %s", self.name, e)
# ------------------------------------------------------------------
# Gen 2: Background dedup maintenance
# ------------------------------------------------------------------
async def _dedup_cleanup_forever(self) -> None:
"""Periodically purge stale entries from the dedup map.
This runs as a background task so the per-message _is_duplicate()
path stays O(1) — no cleanup cost on every message.
Also triggers the health stats callback on each cycle if one is
registered — natural tick for periodic monitoring.
"""
cycle = 0
while self._running:
try:
await asyncio.sleep(DEDUP_CLEANUP_INTERVAL)
self._prune_dedup_map()
cycle += 1
# Report health every N cycles based on callback interval
if (
self._health_callback
and cycle * DEDUP_CLEANUP_INTERVAL
>= self._health_callback_interval
):
await self._report_health_if_callback()
cycle = 0
except asyncio.CancelledError:
break
except Exception:
logger.debug(
"[%s] Dedup cleanup error (non-fatal)", self.name
)
def _prune_dedup_map(self) -> None:
"""Remove entries older than DEDUP_WINDOW from the dedup map."""
now = time.time()
cutoff = now - DEDUP_WINDOW
before = len(self._seen_message_ids)
self._seen_message_ids = {
k: v for k, v in self._seen_message_ids.items() if v > cutoff
}
after = len(self._seen_message_ids)
if before != after:
logger.debug(
"[%s] Dedup map pruned: %d -> %d entries",
self.name, before, after,
)
# ------------------------------------------------------------------
# Gen 3: Periodic @all-bots refresh
# ------------------------------------------------------------------
async def _all_bots_refresh_forever(self) -> None:
"""Periodically re-resolve the @all-bots user ID.
Runs every ALL_BOTS_REFRESH_INTERVAL (1 hour by default).
Handles the case where the @all-bots user is created or
recreated after initial connection.
"""
while self._running:
try:
await asyncio.sleep(ALL_BOTS_REFRESH_INTERVAL)
await self._resolve_all_bots_user_id()
except asyncio.CancelledError:
break
except Exception:
logger.debug(
"[%s] @all-bots refresh error (non-fatal)", self.name
)
# ------------------------------------------------------------------
# Gen 2: Dynamic @all-bots resolution
# ------------------------------------------------------------------
async def _resolve_all_bots_user_id(self) -> None:
"""Try to resolve the @all-bots user ID from the Zulip server.
Falls back to the configured or default value if resolution fails.
"""
try:
resp = await self._api_call("GET", "/api/v1/users")
if not resp:
return
members = resp.get("members", [])
for member in members:
email = member.get("email", "")
# @all-bots typically has an email pattern like all-bots@...
if "all-bots" in email.lower():
uid = member.get("user_id")
if uid is not None:
self._all_bots_user_id = int(uid)
self._health_stats["all_bots_refreshes"] += 1
logger.info(
"[%s] Resolved @all-bots user_id=%s from %s",
self.name, uid, email,
)
return
except Exception as e:
logger.debug(
"[%s] Could not resolve @all-bots ID dynamically: %s. "
"Using configured default %s.",
self.name, e, self._all_bots_user_id,
)
# ------------------------------------------------------------------
# Message processing — Zulip event -> MessageEvent -> Gateway
# ------------------------------------------------------------------
async def _process_zulip_event(self, event: Dict[str, Any]) -> None:
"""Convert a Zulip message event to MessageEvent and dispatch to gateway."""
msg = event.get("message", {})
msg_type = msg.get("type", "") # "private" or "stream"
sender_email = msg.get("sender_email", "")
sender_name = msg.get("sender_full_name", "Unknown")
sender_id = msg.get("sender_id")
content = msg.get("content", "")
# Echo-loop prevention: skip own messages
if sender_email == self._bot_email:
return
# Gen 2: Check for self-test response
if self._selftest_future and not self._selftest_future.done():
# Detect if this is a response to our self-test
self._selftest_future.set_result(True)
return
# Deduplication (O(1) — no cleanup on this path)
msg_id = f"{event.get('id', '')}:{msg.get('id', '')}"
if self._is_duplicate(msg_id):
return
# Update health stats
self._health_stats["messages_received"] += 1
self._health_stats["last_message_at"] = datetime.now(timezone.utc).isoformat()
# Determine if this message targets this bot
mentioned_users = msg.get("mentioned_users", []) or []
mentioned_ids = [u.get("user_id") for u in mentioned_users if isinstance(u, dict)]
is_dm = msg_type == "private"
is_direct_mention = self._bot_user_id and self._bot_user_id in mentioned_ids
is_all_bots = self._all_bots_user_id in mentioned_ids
# DM-first: process all private messages
if is_dm:
self._health_stats["dms_routed"] += 1
logger.info("[%s] DM from %s: %.60s", self.name, sender_name, content)
await self._route_message(
text=content,
chat_id=str(sender_id),
chat_name=sender_name,
user_id=str(sender_id),
user_name=sender_name,
chat_type="dm",
msg_id=msg_id,
timestamp=msg.get("timestamp"),
raw=event,
)
return
# Stream: only respond to @mentions and @all-bots
if msg_type == "stream" and (is_direct_mention or is_all_bots):
self._health_stats["mentions_routed"] += 1
stream_name = msg.get("display_recipient", "unknown")
topic = msg.get("subject", "general")
logger.info(
"[%s] %s in #%s > %s: %.60s",
self.name,
"@mention" if is_direct_mention else "@all-bots",
stream_name, topic, content,
)
# Clean @mention artifacts from content
clean_content = MENTION_CLEANER.sub("", content).strip()
chat_id = f"{stream_name}:{topic}"
await self._route_message(
text=clean_content,
chat_id=chat_id,
chat_name=f"#{stream_name} > {topic}",
user_id=str(sender_id),
user_name=sender_name,
chat_type="stream",
thread_id=topic,
chat_topic=topic,
msg_id=msg_id,
timestamp=msg.get("timestamp"),
raw=event,
reply_to_id=msg.get("id"),
reply_to_type="stream",
reply_to_stream=stream_name,
reply_to_topic=topic,
)
async def _route_message(
self,
text: str,
chat_id: str,
chat_name: str,
user_id: str,
user_name: str,
chat_type: str,
msg_id: str,
timestamp: Any,
raw: Dict[str, Any],
thread_id: Optional[str] = None,
chat_topic: Optional[str] = None,
reply_to_id: Optional[int] = None,
reply_to_type: Optional[str] = None,
reply_to_stream: Optional[str] = None,
reply_to_topic: Optional[str] = None,
) -> None:
"""Build MessageEvent and send to gateway via handle_message."""
source = self.build_source(
chat_id=chat_id,
chat_name=chat_name,
chat_type=chat_type,
user_id=user_id,
user_name=user_name,
thread_id=thread_id,
chat_topic=chat_topic,
chat_id_alt=str(reply_to_id) if reply_to_id else None,
)
# Derive message type
message_type = MessageType.TEXT
message_event = MessageEvent(
text=text,
message_type=message_type,
source=source,
message_id=msg_id,
raw_message=raw,
timestamp=_parse_zulip_timestamp(timestamp),
)
# Store reply metadata so send() can respond to the right thread
if reply_to_id:
self._pending_replies[chat_id] = reply_to_id
logger.debug("[%s] Dispatching message from %s", self.name, user_name)
await self.handle_message(message_event)
# ------------------------------------------------------------------
# Outbound messaging — send responses to Zulip
# ------------------------------------------------------------------
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a message to a Zulip chat.
For DMs, chat_id is the user_id. For streams, format is
"stream_name:topic". Sends a placeholder first if this is the
first message in a turn, then edits it on subsequent calls.
"""
# Determine target type and IDs
is_dm = not chat_id or ":" not in chat_id
truncated = _truncate(content)
if is_dm:
# DM: chat_id is the user_id
to = [int(chat_id)]
payload = {
"type": "private",
"to": json.dumps(to),
"content": truncated,
}
else:
# Stream: chat_id is "stream_name:topic"
parts = chat_id.split(":", 1)
stream_name = parts[0]
topic = parts[1] if len(parts) > 1 else "general"
payload = {
"type": "stream",
"to": stream_name,
"subject": topic,
"content": truncated,
}
# Check if we should send a placeholder first (first message in turn)
placeholder_msg_id = self._pending_replies.pop(chat_id, None)
if placeholder_msg_id:
# This is the first response — send placeholder and return its ID
placeholder = self._pick_placeholder()
placehold_payload = {**payload, "content": placeholder}
result = await self._send_api_call(placehold_payload)
if result and result.get("id"):
msg_id = str(result["id"])
logger.debug(
"[%s] Sent placeholder msg=%s for %s",
self.name, msg_id, chat_id,
)
self._health_stats["send_count"] += 1
return SendResult(
success=True,
platform="zulip",
chat_id=chat_id,
message_id=msg_id,
metadata={"placeholder": True},
)
# Send actual content
result = await self._send_api_call(payload)
if result and result.get("id"):
self._health_stats["send_count"] += 1
return SendResult(
success=True,
platform="zulip",
chat_id=chat_id,
message_id=str(result["id"]),
)
self._health_stats["send_errors"] += 1
return SendResult(
success=False,
platform="zulip",
chat_id=chat_id,
error="Failed to send message",
)
async def edit_message(
self,
chat_id: str,
message_id: str,
content: str,
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""Edit a previously sent Zulip message (for placeholder replacement)."""
truncated = _truncate(content)
payload = {
"message_id": int(message_id),
"content": truncated,
}
try:
resp = await self._api_call("PATCH", "/api/v1/messages", data=payload)
return resp is not None
except Exception as e:
logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e)
return False
async def send_typing(self, chat_id: str, metadata: Optional[Dict] = None) -> None:
"""Send typing indicator to a Zulip chat."""
if ":" in chat_id:
return # No typing for stream messages
try:
user_ids = json.dumps([int(chat_id)])
await self._api_call(
"POST", "/api/v1/typing",
data={"to": user_ids, "op": "start"},
)
except Exception:
pass # Non-critical
async def stop_typing(self, chat_id: str) -> None:
"""Stop typing indicator."""
if ":" in chat_id:
return
try:
user_ids = json.dumps([int(chat_id)])
await self._api_call(
"POST", "/api/v1/typing",
data={"to": user_ids, "op": "stop"},
)
except Exception:
pass # Non-critical
async def get_chat_info(self, chat_id: str) -> dict:
"""Return basic info about a Zulip chat."""
if ":" in chat_id:
parts = chat_id.split(":", 1)
return {"name": parts[0], "type": "channel", "topic": parts[1]}
return {"name": chat_id, "type": "dm"}
async def delete_message(
self, chat_id: str, message_id: str, metadata: Optional[Dict] = None
) -> bool:
"""Delete a message via Zulip API."""
# Zulip doesn't support message deletion via API for bots
# Send an empty edit instead
return await self.edit_message(chat_id, message_id, "*deleted*")
# ------------------------------------------------------------------
# Gen 2: Self-test diagnostics
# ------------------------------------------------------------------
async def selftest(self) -> Dict[str, Any]:
"""Run a self-test to verify adapter health.
Checks:
1. Connection state (queue registered, HTTP client alive)
2. Bot identity resolved (bot_user_id is set)
3. Queue polling active (poll_task is running)
4. Echo-loop prevention configured (bot_email matches registered email)
5. Subscription to primary stream (if reachable)
6. @all-bots ID resolved
7. Dedup maintenance task running
Returns a dict with pass/fail for each check and an overall verdict.
"""
checks = {}
# 1. Connection state
checks["connected"] = {
"status": self._connected,
"detail": "Queue registered and connected"
if self._connected
else "Not connected to Zulip",
}
# 2. Queue registered
checks["queue_registered"] = {
"status": self._queue_id is not None,
"detail": f"Queue: {self._queue_id}"
if self._queue_id
else "No queue registered",
}
# 3. HTTP client alive
checks["http_client"] = {
"status": self._http_client is not None and not self._http_client.is_closed,
"detail": "HTTP client active"
if self._http_client and not self._http_client.is_closed
else "HTTP client unavailable",
}
# 4. Bot identity
checks["bot_identity"] = {
"status": self._bot_user_id is not None,
"detail": f"Bot user_id: {self._bot_user_id}"
if self._bot_user_id
else "Bot user_id not resolved",
}
# 5. Poll loop active
checks["poll_loop"] = {
"status": self._poll_task is not None and not self._poll_task.done(),
"detail": "Poll loop running"
if self._poll_task and not self._poll_task.done()
else "Poll loop not active",
}
# 6. Dedup cleanup active
checks["dedup_cleanup"] = {
"status": self._dedup_cleanup_task is not None
and not self._dedup_cleanup_task.done(),
"detail": "Dedup maintenance running"
if self._dedup_cleanup_task and not self._dedup_cleanup_task.done()
else "Dedup maintenance not active",
}
# 7. Echo-loop prevention
checks["echo_prevention"] = {
"status": bool(self._bot_email),
"detail": f"Bot email: {self._bot_email}",
}
# 8. @all-bots configured
checks["all_bots_configured"] = {
"status": self._all_bots_user_id is not None,
"detail": f"@all-bots user_id: {self._all_bots_user_id}",
}
# Overall verdict
critical = ["connected", "queue_registered", "http_client", "poll_loop"]
passed = sum(1 for c in checks.values() if c["status"])
critical_passed = sum(1 for k in critical if checks.get(k, {}).get("status"))
failed = [k for k, v in checks.items() if not v["status"]]
if critical_passed == len(critical) and passed == len(checks):
verdict = "healthy"
elif critical_passed < len(critical):
verdict = "critical_failure"
else:
verdict = "degraded"
return {
"verdict": verdict,
"timestamp": datetime.now(timezone.utc).isoformat(),
"checks": checks,
"summary": {
"total": len(checks),
"passed": passed,
"failed": len(failed),
"failed_checks": failed,
"critical_passed": critical_passed,
"critical_total": len(critical),
},
}
# ------------------------------------------------------------------
# Gen 3: Health stats callback
# ------------------------------------------------------------------
def set_health_callback(
self,
callback,
interval: int = 600,
) -> None:
"""Register a callback for periodic health stats reporting.
The callback is called with health stats dict every `interval`
seconds. Can be used by the Hermes Gateway to log health to
RA-H OS knowledge graph or external monitoring.
Args:
callback: Async callable receiving health stats dict
interval: Seconds between calls (default 600 = 10 min)
"""
self._health_callback = callback
self._health_callback_interval = interval
async def _report_health_if_callback(self) -> None:
"""Call the health stats callback if one is registered."""
if self._health_callback:
try:
stats = await self.get_health_stats()
await self._health_callback(stats)
except Exception as e:
logger.debug(
"[%s] Health callback error: %s", self.name, e
)
# ------------------------------------------------------------------
# Gen 2: Health stats
# ------------------------------------------------------------------
async def get_health_stats(self) -> Dict[str, Any]:
"""Return health and performance statistics.
Returns a snapshot of:
- Uptime
- Poll counts, errors, reconnects
- Message counts (total, DM, mention)
- Send counts and errors
- Dedup map size
- Timestamps of last activity
Suitable for periodic logging to RA-H OS knowledge graph.
"""
now = datetime.now(timezone.utc)
uptime = None
if self._health_stats.get("started_at"):
try:
started = datetime.fromisoformat(self._health_stats["started_at"])
uptime_seconds = (now - started).total_seconds()
uptime = f"{uptime_seconds:.0f}s"
except (ValueError, TypeError):
pass
stats = dict(self._health_stats)
stats["uptime"] = uptime
stats["dedup_map_size"] = len(self._seen_message_ids)
stats["queue_id"] = self._queue_id
stats["bot_user_id"] = self._bot_user_id
stats["all_bots_user_id"] = self._all_bots_user_id
stats["connected"] = self._connected
stats["checked_at"] = now.isoformat()
# Compute error rate
total_polls = stats.get("poll_count", 0) or 1
stats["error_rate"] = round(
stats.get("poll_errors", 0) / total_polls, 4
)
return stats
# ------------------------------------------------------------------
# Zulip API helpers
# ------------------------------------------------------------------
async def _api_call(
self,
method: str,
path: str,
data: Optional[Dict] = None,
params: Optional[Dict] = None,
) -> Optional[Dict]:
"""Make an API call to Zulip."""
if not self._http_client:
return None
url = f"{self._site}{path}"
headers = {
"Authorization": self._auth_header,
}
try:
if method == "GET":
response = await self._http_client.get(
url, params=params, headers=headers,
)
elif method == "POST":
response = await self._http_client.post(
url, data=data, headers=headers,
)
elif method == "PATCH":
response = await self._http_client.patch(
url, data=data, headers=headers,
)
else:
return None
if response.status_code >= 400:
logger.warning(
"[%s] API %s %s: %d %s",
self.name, method, path,
response.status_code, response.text[:200],
)
return None
return response.json()
except httpx.TimeoutException:
logger.debug("[%s] Timeout on %s %s", self.name, method, path)
return None
except Exception as e:
logger.warning("[%s] API error %s %s: %s", self.name, method, path, e)
return None
async def _send_api_call(self, payload: Dict) -> Optional[Dict]:
"""Send a message to Zulip."""
return await self._api_call("POST", "/api/v1/messages", data=payload)
# ------------------------------------------------------------------
# Utilities
# ------------------------------------------------------------------
def _pick_placeholder(self) -> str:
"""Pick a random placeholder message for streaming UX."""
idx = hash(str(time.time())) % len(PLACEHOLDERS)
return PLACEHOLDERS[idx]
def _is_duplicate(self, msg_id: str) -> bool:
"""Deduplication using message IDs — O(1) lookup, no cleanup.
Cleanup runs in a separate background task (_dedup_cleanup_forever)
so this path stays fast.
"""
if msg_id in self._seen_message_ids:
return True
self._seen_message_ids[msg_id] = time.time()
return False
def get_all_bots_user_id(self) -> int:
"""Return the current @all-bots user ID.
This is used by the zulip-mention-reliability contract for
cross-contract verification.
"""
return self._all_bots_user_id
def get_bot_user_id(self) -> Optional[int]:
"""Return the bot's own user ID for identity checks."""
return self._bot_user_id
# ------------------------------------------------------------------
# Plugin registration helpers
# ------------------------------------------------------------------
def check_requirements() -> bool:
"""Check whether the Zulip adapter is installable."""
if not HTTPX_AVAILABLE:
return False
site = os.getenv("ZULIP_SITE", "").strip()
email = os.getenv("ZULIP_EMAIL", "").strip()
api_key = os.getenv("ZULIP_API_KEY", "").strip()
return bool(site and email and api_key)
def validate_config(config) -> bool:
"""Validate that the configured Zulip platform has credentials."""
extra = getattr(config, "extra", {}) or {}
site = extra.get("site") or os.getenv("ZULIP_SITE", "")
email = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
api_key = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")
return bool(site and email and api_key)
def is_connected(config) -> bool:
"""Check whether Zulip is configured (env or config.yaml)."""
return validate_config(config)
def _env_enablement() -> Optional[dict]:
"""Seeds PlatformConfig.extra from env vars for env-only setups."""
site = os.getenv("ZULIP_SITE", "").strip()
email = os.getenv("ZULIP_EMAIL", "").strip()
api_key = os.getenv("ZULIP_API_KEY", "").strip()
if not (site and email and api_key):
return None
result: Dict[str, Any] = {
"site": site,
"email": email,
"api_key": api_key,
}
stream = os.getenv("ZULIP_STREAM", "").strip()
if stream:
result["stream"] = stream
agent_name = os.getenv("ZULIP_AGENT_NAME", "").strip()
if agent_name:
result["agent_name"] = agent_name
home = os.getenv("ZULIP_HOME_CHANNEL", "").strip()
if home:
result["home_channel"] = home
return result
def register(ctx) -> None:
"""Plugin entry point — called by the Hermes plugin system at startup."""
ctx.register_platform(
name="zulip",
label="Zulip",
adapter_factory=lambda cfg: ZulipAdapter(cfg),
check_fn=check_requirements,
validate_config=validate_config,
is_connected=is_connected,
required_env=["ZULIP_SITE", "ZULIP_EMAIL", "ZULIP_API_KEY"],
install_hint="pip install httpx # already a Hermes dependency",
env_enablement_fn=_env_enablement,
cron_deliver_env_var="ZULIP_HOME_CHANNEL",
allowed_users_env="ZULIP_ALLOWED_USERS",
allow_all_env="ZULIP_ALLOW_ALL_USERS",
max_message_length=MAX_ZULIP_MESSAGE,
emoji="💬",
pii_safe=True,
allow_update_command=True,
platform_hint=(
"You are communicating via Zulip. Format responses with "
"Zulip-compatible Markdown. Keep responses concise and "
"well-structured. DMs are private conversations; stream "
"messages are visible to all subscribers."
),
)
Reference in New Issue View Git Blame Copy Permalink