Abiba (pi)
909b9ce029
- add_known_issue() / get_known_issues() / clear_known_issues() for cross-generation tracking - simulate_dm_latency() for offline dm_response_time_ms < 10000 verification - _pick_placeholder() switched from hash-based to random.choice() - Docstring corrected: ordered Gen 6→5→4→3→2 with proper attribution - known_issues exported in get_health_stats() for KG logging
1609 lines
60 KiB
Python
1609 lines
60 KiB
Python
"""
|
|
Zulip platform adapter (Hermes plugin) — Gen 6.
|
|
|
|
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 6 Improvements (2026-07-07):
|
|
1. known_issues persistence — add_known_issue(), get_known_issues(),
|
|
clear_known_issues() methods. Issues tracked in health stats for
|
|
external logging to knowledge graph across generations.
|
|
2. simulate_dm_latency() — offline-verifiable DM processing pipeline
|
|
timing test. Measures event parse, placeholder pick, auth, dedup,
|
|
and truncation to verify dm_response_time_ms < 10000 without live Zulip.
|
|
3. Placeholder collision fix — switched from hash-based pick to
|
|
random.choice() to avoid identical placeholders on rapid-fire calls.
|
|
|
|
Gen 5 Improvements (2026-06-29):
|
|
1. Bot user ID resolution — fetches from /users/me when /register doesn't return it
|
|
2. Stream subscription auto-fix — ensures bot is subscribed to primary streams
|
|
on connect, critical for receiving @mentions in stream topics
|
|
|
|
Gen 4 Improvements (2026-06-27):
|
|
1. HTTP client recreation on reconnect — new httpx.AsyncClient() after queue expiry
|
|
or sustained failures to prevent CLOSE-WAIT socket leaks
|
|
2. Heartbeat logging — periodic "still alive" message to prove poll loop is running
|
|
3. Log sanitization — truncates 502/error HTML to first 80 chars (no Netbird HTML spam)
|
|
4. Connection recovery — creates fresh HTTP client after N empty poll cycles
|
|
5. Poll silence detection — logs warnings when no events received for extended period
|
|
6. Connection reuse limit — periodic client recreation to prevent connection pool exhaustion
|
|
|
|
Gen 3 Improvements (2026-06-26):
|
|
1. Malformed message resilience — try/except wraps each event, no crash on bad JSON
|
|
2. Periodic @all-bots refresh — background task re-resolves user ID every hour
|
|
3. Health stats callback — external consumers can register for periodic stats
|
|
|
|
Gen 2 Improvements (2026-06-26):
|
|
1. Background dedup maintenance — O(1) per-message path, periodic cleanup task
|
|
2. Self-test capability — selftest() method verifying DM/mention/queue health
|
|
3. Health stats tracking — poll stats, reconnect counts, error rates
|
|
4. Dynamic @all-bots resolution — falls back to hardcoded default
|
|
|
|
Architecture (mirrors pi-zulip-extension):
|
|
Zulip event queue -> poll loop -> parse message -> handle_message(MessageEvent)
|
|
-> Gateway processes -> send() / edit_message() posts response to Zulip
|
|
|
|
No external SDK required — only httpx (already a Hermes dependency).
|
|
Ships as a Hermes platform plugin under plugins/platforms/zulip/.
|
|
"""
|
|
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import re
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
try:
|
|
import httpx
|
|
HTTPX_AVAILABLE = True
|
|
except ImportError:
|
|
HTTPX_AVAILABLE = False
|
|
httpx = None
|
|
|
|
from gateway.config import Platform, PlatformConfig
|
|
from gateway.platforms.base import (
|
|
BasePlatformAdapter,
|
|
MessageEvent,
|
|
MessageType,
|
|
SendResult,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Constants
|
|
DEFAULT_STREAM = "agent-hub"
|
|
DEFAULT_ALL_BOTS_USER_ID = 1
|
|
DEFAULT_POLL_INTERVAL = 3.0
|
|
MAX_ZULIP_MESSAGE = 10000
|
|
ECHO_TAG_PREFIX = "hermes-zulip-"
|
|
RECONNECT_BACKOFF = [2, 5, 10, 30, 60]
|
|
DEDUP_WINDOW = 300 # 5 minutes
|
|
DEDUP_MAX_SIZE = 1000
|
|
DEDUP_CLEANUP_INTERVAL = 120 # Clean old entries every 2 minutes
|
|
ALL_BOTS_REFRESH_INTERVAL = 3600 # Re-resolve @all-bots user ID every hour
|
|
SELFTEST_TIMEOUT = 30 # seconds to wait for self-test response
|
|
|
|
# Gen 4: Connection pool health
|
|
HEARTBEAT_INTERVAL = 300 # Log heartbeat every 5 minutes
|
|
MAX_CONSECUTIVE_EMPTY_POLLS = 500 # Reset client after this many empty polls (~25min at 3s interval). Raised from 20 to prevent idle bots from cycling reconnections.
|
|
CLIENT_REUSE_LIMIT = 500 # Create fresh client every N polls to prevent connection leaks
|
|
POLL_SILENCE_WARN_INTERVAL = 60 # Warn if no events received for this many seconds
|
|
MAX_ERROR_LOG_LEN = 80 # Truncate error response bodies to avoid HTML spam
|
|
|
|
# Regex to strip Zulip @mention markup
|
|
MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*")
|
|
|
|
# Placeholder messages for streaming UX
|
|
PLACEHOLDERS = [
|
|
":robot: _Processing your message..._",
|
|
":hourglass_flowing_sand: _Thinking..._",
|
|
":brain: _Generating response..._",
|
|
]
|
|
|
|
|
|
def _build_auth_header(email: str, api_key: str) -> str:
|
|
"""Build Basic auth header for Zulip API."""
|
|
token = base64.b64encode(f"{email}:{api_key}".encode()).decode()
|
|
return f"Basic {token}"
|
|
|
|
|
|
def _truncate(text: str, limit: int = MAX_ZULIP_MESSAGE) -> str:
|
|
"""Truncate to Zulip's message limit with notice."""
|
|
if len(text) <= limit:
|
|
return text
|
|
return text[:limit] + "\n\n[...truncated at Zulip limit]"
|
|
|
|
|
|
def _parse_zulip_timestamp(ts: Any) -> datetime:
|
|
"""Convert Zulip timestamp (float seconds) to datetime."""
|
|
try:
|
|
return datetime.fromtimestamp(float(ts), tz=timezone.utc)
|
|
except (ValueError, OSError, TypeError):
|
|
return datetime.now(tz=timezone.utc)
|
|
|
|
|
|
class ZulipAdapter(BasePlatformAdapter):
|
|
"""Zulip adapter for Hermes agents.
|
|
|
|
Connects to Zulip via event queues, polls for new events, converts
|
|
DMs and @mentions to MessageEvent, and sends responses with streaming
|
|
placeholder->edit UX.
|
|
|
|
Gen 3 adds:
|
|
- Malformed message resilience (per-event try/except)
|
|
- Periodic @all-bots refresh (hourly background task)
|
|
- Health stats callback for external consumers
|
|
|
|
Gen 2 adds:
|
|
- Background dedup maintenance task (O(1) per-message)
|
|
- Self-test diagnostics (DM/mention/queue health)
|
|
- Health stats for monitoring
|
|
- Dynamic @all-bots resolution
|
|
"""
|
|
|
|
def __init__(self, config: PlatformConfig):
|
|
platform = Platform("zulip")
|
|
super().__init__(config=config, platform=platform)
|
|
|
|
extra = config.extra or {}
|
|
|
|
# Zulip connection
|
|
self._site: str = (
|
|
extra.get("site") or os.getenv("ZULIP_SITE", "")
|
|
).rstrip("/")
|
|
self._email: str = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
|
|
self._api_key: str = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")
|
|
|
|
# Agent identity
|
|
self._agent_name: str = (
|
|
extra.get("agent_name") or os.getenv("ZULIP_AGENT_NAME", "hermes-agent")
|
|
)
|
|
|
|
# Stream config
|
|
self._stream: str = (
|
|
extra.get("stream") or os.getenv("ZULIP_STREAM", DEFAULT_STREAM)
|
|
)
|
|
|
|
# Self-test config — owner user ID for DM self-tests
|
|
self._owner_user_id: Optional[int] = self._resolve_int_opt(
|
|
extra.get("owner_user_id") or os.getenv("ZULIP_OWNER_USER_ID", "")
|
|
)
|
|
|
|
# @all-bots user ID — dynamically resolved on connect, falls back to config/env/default
|
|
self._all_bots_user_id: int = self._resolve_int(
|
|
extra.get("all_bots_user_id")
|
|
or os.getenv("ZULIP_ALL_BOTS_USER_ID", str(DEFAULT_ALL_BOTS_USER_ID)),
|
|
DEFAULT_ALL_BOTS_USER_ID,
|
|
)
|
|
|
|
# Polling
|
|
self._poll_interval: float = float(
|
|
extra.get("poll_interval")
|
|
or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL))
|
|
)
|
|
|
|
# --- Gen 4: Connection health ---
|
|
self._consecutive_empty_polls: int = 0
|
|
self._total_poll_count: int = 0
|
|
self._last_event_received_time: float = time.time()
|
|
self._last_heartbeat_time: float = 0.0
|
|
self._client_pool_reset_count: int = 0
|
|
|
|
# --- State ---
|
|
# Support both ZULIP_SITE and ZULIP_URL env vars
|
|
if not self._site:
|
|
self._site = (os.getenv("ZULIP_URL", "") or "").rstrip("/")
|
|
|
|
self._auth_header: str = _build_auth_header(self._email, self._api_key)
|
|
self._queue_id: Optional[str] = None
|
|
self._last_event_id: int = -1
|
|
self._poll_task: Optional[asyncio.Task] = None
|
|
self._dedup_cleanup_task: Optional[asyncio.Task] = None
|
|
self._http_client: Optional[httpx.AsyncClient] = None
|
|
self._pending_replies: Dict[str, int] = {} # chat_id -> placeholder zulip_msg_id
|
|
|
|
# Derived identity for echo-loop prevention
|
|
self._bot_user_id: Optional[int] = None
|
|
self._bot_email: str = self._email
|
|
|
|
# --- Gen 2: Dedup (background-maintained) ---
|
|
self._seen_message_ids: Dict[str, float] = {}
|
|
|
|
# --- Gen 2/3: Health stats ---
|
|
self._health_stats: Dict[str, Any] = {
|
|
"started_at": None,
|
|
"poll_count": 0,
|
|
"poll_errors": 0,
|
|
"malformed_events": 0,
|
|
"messages_received": 0,
|
|
"dms_routed": 0,
|
|
"mentions_routed": 0,
|
|
"reconnects": 0,
|
|
"send_count": 0,
|
|
"send_errors": 0,
|
|
"all_bots_refreshes": 0,
|
|
"last_poll_at": None,
|
|
"last_message_at": None,
|
|
"last_error_at": None,
|
|
"last_error_msg": None,
|
|
}
|
|
|
|
# --- Gen 2: Self-test future for async response ---
|
|
self._selftest_future: Optional[asyncio.Future] = None
|
|
|
|
# --- Gen 3: Periodic @all-bots refresh ---
|
|
self._all_bots_refresh_task: Optional[asyncio.Task] = None
|
|
|
|
# --- Gen 6: known_issues persistence ---
|
|
self._known_issues: List[str] = []
|
|
|
|
# --- Gen 3: Health stats callback ---
|
|
self._health_callback = None
|
|
self._health_callback_interval: int = 300 # 5 min
|
|
|
|
@staticmethod
|
|
def _resolve_int(val: Any, default: int) -> int:
|
|
"""Resolve an integer config value, falling back to default."""
|
|
if val is None or val == "":
|
|
return default
|
|
try:
|
|
return int(val)
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
@staticmethod
|
|
def _resolve_int_opt(val: Any) -> Optional[int]:
|
|
"""Resolve an optional integer config value."""
|
|
if val is None or val == "":
|
|
return None
|
|
try:
|
|
return int(val)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Connection lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
async def connect(self, **kwargs: Any) -> bool:
|
|
"""Connect to Zulip and register an event queue.
|
|
|
|
Accepts **kwargs for Hermes Gateway compatibility (e.g., is_reconnect).
|
|
"""
|
|
if not HTTPX_AVAILABLE:
|
|
logger.warning("[%s] httpx not installed", self.name)
|
|
return False
|
|
|
|
if not self._site or not self._email or not self._api_key:
|
|
logger.warning("[%s] Zulip credentials not configured", self.name)
|
|
return False
|
|
|
|
try:
|
|
await self._create_http_client()
|
|
self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Reset connection health counters
|
|
self._consecutive_empty_polls = 0
|
|
self._total_poll_count = 0
|
|
self._last_event_received_time = time.time()
|
|
|
|
# Register event queue
|
|
queue_resp, _ = await self._api_call(
|
|
"POST", "/api/v1/register",
|
|
data={
|
|
"event_types": '["message"]',
|
|
"apply_markdown": "true",
|
|
"include_subscribers": "false",
|
|
},
|
|
)
|
|
if not queue_resp:
|
|
logger.error("[%s] Failed to register event queue", self.name)
|
|
return False
|
|
|
|
data = queue_resp
|
|
self._queue_id = data.get("queue_id")
|
|
self._last_event_id = data.get("last_event_id", -1)
|
|
self._bot_user_id = data.get("user_id")
|
|
|
|
# Resolve bot user_id if register endpoint didn't provide it
|
|
# (common on some Zulip versions)
|
|
if self._bot_user_id is None:
|
|
await self._resolve_bot_user_id()
|
|
|
|
# Gen 2: Try to resolve @all-bots user ID dynamically
|
|
await self._resolve_all_bots_user_id()
|
|
|
|
# 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)",
|
|
self.name, self._site, self._email,
|
|
self._queue_id, self._bot_user_id, self._all_bots_user_id,
|
|
)
|
|
|
|
# Start poll loop
|
|
self._poll_task = asyncio.create_task(self._poll_forever())
|
|
|
|
# Gen 2: Start background dedup cleanup task
|
|
self._dedup_cleanup_task = asyncio.create_task(
|
|
self._dedup_cleanup_forever()
|
|
)
|
|
|
|
# Gen 3: Start periodic @all-bots refresh task
|
|
self._all_bots_refresh_task = asyncio.create_task(
|
|
self._all_bots_refresh_forever()
|
|
)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error("[%s] Connection failed: %s", self.name, e)
|
|
return False
|
|
|
|
async def disconnect(self) -> None:
|
|
"""Disconnect from Zulip and cancel all background tasks."""
|
|
# Cancel dedup cleanup task
|
|
if self._dedup_cleanup_task:
|
|
self._dedup_cleanup_task.cancel()
|
|
try:
|
|
await self._dedup_cleanup_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._dedup_cleanup_task = None
|
|
|
|
# Cancel @all-bots refresh task
|
|
if self._all_bots_refresh_task:
|
|
self._all_bots_refresh_task.cancel()
|
|
try:
|
|
await self._all_bots_refresh_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._all_bots_refresh_task = None
|
|
|
|
# Cancel poll task
|
|
if self._poll_task:
|
|
self._poll_task.cancel()
|
|
try:
|
|
await self._poll_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._poll_task = None
|
|
|
|
self._queue_id = None
|
|
self._mark_disconnected()
|
|
logger.info("[%s] Disconnected", self.name)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Poll loop — receives messages from Zulip
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _poll_forever(self) -> None:
|
|
"""Continuously poll the Zulip event queue."""
|
|
backoff_idx = 0
|
|
|
|
while self._running:
|
|
if not self._queue_id:
|
|
await asyncio.sleep(self._poll_interval)
|
|
continue
|
|
|
|
try:
|
|
events = await self._fetch_events()
|
|
self._health_stats["poll_count"] += 1
|
|
self._total_poll_count += 1
|
|
self._health_stats["last_poll_at"] = datetime.now(
|
|
timezone.utc
|
|
).isoformat()
|
|
|
|
if events:
|
|
now = time.time()
|
|
self._consecutive_empty_polls = 0
|
|
self._last_event_received_time = now
|
|
for event in events:
|
|
try:
|
|
await self._process_zulip_event(event)
|
|
except Exception as e:
|
|
# Gen 3: Malformed message resilience — catch per-event
|
|
# failures so one bad message never kills the poll loop
|
|
logger.warning(
|
|
"[%s] Malformed event skipped: %s. "
|
|
"Event id=%s",
|
|
self.name, e,
|
|
event.get("id", "unknown"),
|
|
)
|
|
self._health_stats["malformed_events"] = (
|
|
self._health_stats.get("malformed_events", 0) + 1
|
|
)
|
|
else:
|
|
# Gen 4: Track consecutive empty polls to detect silent failures
|
|
self._consecutive_empty_polls += 1
|
|
|
|
# Gen 4: Detect prolonged silence (no events for too long)
|
|
silence_duration = time.time() - self._last_event_received_time
|
|
if silence_duration > POLL_SILENCE_WARN_INTERVAL:
|
|
logger.warning(
|
|
"[%s] No events received for %.0fs "
|
|
"(consecutive_empty=%d, total_polls=%d, "
|
|
"queue=%s)",
|
|
self.name, silence_duration,
|
|
self._consecutive_empty_polls,
|
|
self._total_poll_count,
|
|
self._queue_id,
|
|
)
|
|
|
|
# Gen 4: Reset HTTP client if too many empty polls in a row
|
|
# (connection pool likely stuck in CLOSE-WAIT)
|
|
if (
|
|
self._consecutive_empty_polls
|
|
>= MAX_CONSECUTIVE_EMPTY_POLLS
|
|
):
|
|
logger.warning(
|
|
"[%s] %d consecutive empty polls — "
|
|
"recreating HTTP client and reconnecting",
|
|
self.name, self._consecutive_empty_polls,
|
|
)
|
|
await self._reconnect_with_fresh_client()
|
|
backoff_idx = 0
|
|
self._consecutive_empty_polls = 0
|
|
continue
|
|
|
|
# Gen 4: Periodic client refresh to prevent connection leaks
|
|
if self._total_poll_count % CLIENT_REUSE_LIMIT == 0:
|
|
logger.info(
|
|
"[%s] Client reuse limit reached (%d polls) — "
|
|
"creating fresh HTTP client",
|
|
self.name, self._total_poll_count,
|
|
)
|
|
old_client = self._http_client
|
|
await self._create_http_client()
|
|
if old_client:
|
|
try:
|
|
await old_client.aclose()
|
|
except Exception:
|
|
pass
|
|
|
|
# Clear stale errors on successful poll cycle
|
|
self._health_stats["last_error_msg"] = None
|
|
self._health_stats["last_error_at"] = None
|
|
|
|
# Gen 4: Heartbeat logging — prove poll loop is alive
|
|
now = time.time()
|
|
if now - self._last_heartbeat_time > HEARTBEAT_INTERVAL:
|
|
self._last_heartbeat_time = now
|
|
silence = now - self._last_event_received_time
|
|
logger.info(
|
|
"[%s] Heartbeat — polls=%d empty=%d "
|
|
"silence=%.0fs errors=%d reconnects=%d "
|
|
"queue=%s",
|
|
self.name, self._total_poll_count,
|
|
self._consecutive_empty_polls, silence,
|
|
self._health_stats["poll_errors"],
|
|
self._health_stats["reconnects"],
|
|
self._queue_id,
|
|
)
|
|
|
|
backoff_idx = 0
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
if not self._running:
|
|
break
|
|
self._health_stats["poll_errors"] += 1
|
|
self._health_stats["last_error_at"] = datetime.now(
|
|
timezone.utc
|
|
).isoformat()
|
|
self._health_stats["last_error_msg"] = str(e)[:200]
|
|
|
|
err_str = str(e)
|
|
if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower():
|
|
logger.info("[%s] Queue expired, re-registering...", self.name)
|
|
await self._reconnect_with_fresh_client()
|
|
backoff_idx = 0
|
|
continue
|
|
logger.warning("[%s] Poll error: %s", self.name, e)
|
|
delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF) - 1)]
|
|
backoff_idx += 1
|
|
await asyncio.sleep(delay)
|
|
|
|
await asyncio.sleep(self._poll_interval)
|
|
|
|
async def _fetch_events(self) -> List[Dict[str, Any]]:
|
|
"""Fetch events from the Zulip event queue.
|
|
|
|
Raises RuntimeError with BAD_EVENT_QUEUE_ID when the queue
|
|
expires, so _poll_forever can reconnect.
|
|
"""
|
|
if not self._queue_id:
|
|
return []
|
|
|
|
resp, raw_status = await self._api_call(
|
|
"GET", "/api/v1/events",
|
|
params={
|
|
"queue_id": self._queue_id,
|
|
"last_event_id": str(self._last_event_id),
|
|
"dont_block": "true",
|
|
},
|
|
)
|
|
# Detect queue expiry from HTTP response
|
|
if raw_status == 400:
|
|
raise RuntimeError("BAD_EVENT_QUEUE_ID: queue expired")
|
|
if not resp:
|
|
return []
|
|
|
|
events = resp.get("events", [])
|
|
# Track last event ID for cursor advancement
|
|
for event in events:
|
|
eid = event.get("id", 0)
|
|
if eid > self._last_event_id:
|
|
self._last_event_id = eid
|
|
|
|
return [e for e in events if e.get("type") == "message"]
|
|
|
|
async def _create_http_client(self) -> None:
|
|
"""Create a fresh httpx client, closing the old one if it exists.
|
|
|
|
Gen 4: This ensures a clean connection pool after sustained failures
|
|
or queue expiry, preventing CLOSE-WAIT socket leaks.
|
|
"""
|
|
old_client = self._http_client
|
|
self._http_client = httpx.AsyncClient(
|
|
timeout=httpx.Timeout(30.0, connect=15.0),
|
|
limits=httpx.Limits(
|
|
max_keepalive_connections=5,
|
|
max_connections=10,
|
|
keepalive_expiry=60.0,
|
|
),
|
|
)
|
|
self._client_pool_reset_count += 1
|
|
if old_client:
|
|
try:
|
|
await old_client.aclose()
|
|
except Exception:
|
|
pass
|
|
|
|
async def _reconnect(self) -> None:
|
|
"""Re-register the event queue."""
|
|
self._queue_id = None
|
|
try:
|
|
resp, _ = await self._api_call(
|
|
"POST", "/api/v1/register",
|
|
data={
|
|
"event_types": '["message"]',
|
|
"apply_markdown": "true",
|
|
"include_subscribers": "false",
|
|
},
|
|
)
|
|
if resp:
|
|
self._queue_id = resp.get("queue_id")
|
|
self._last_event_id = resp.get("last_event_id", -1)
|
|
self._health_stats["reconnects"] += 1
|
|
logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id)
|
|
self._consecutive_empty_polls = 0
|
|
self._last_event_received_time = time.time()
|
|
except Exception as e:
|
|
logger.error("[%s] Reconnect failed: %s", self.name, e)
|
|
|
|
async def _reconnect_with_fresh_client(self) -> None:
|
|
"""Re-register the event queue with a fresh HTTP client.
|
|
|
|
Gen 4: Creates a new httpx client to break out of stuck connection
|
|
pools (CLOSE-WAIT sockets from previous failures).
|
|
"""
|
|
logger.info("[%s] Reconnecting with fresh HTTP client", self.name)
|
|
await self._create_http_client()
|
|
await self._reconnect()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 2: Background dedup maintenance
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _dedup_cleanup_forever(self) -> None:
|
|
"""Periodically purge stale entries from the dedup map.
|
|
|
|
This runs as a background task so the per-message _is_duplicate()
|
|
path stays O(1) — no cleanup cost on every message.
|
|
|
|
Also triggers the health stats callback on each cycle if one is
|
|
registered — natural tick for periodic monitoring.
|
|
"""
|
|
cycle = 0
|
|
while self._running:
|
|
try:
|
|
await asyncio.sleep(DEDUP_CLEANUP_INTERVAL)
|
|
self._prune_dedup_map()
|
|
cycle += 1
|
|
# Report health every N cycles based on callback interval
|
|
if (
|
|
self._health_callback
|
|
and cycle * DEDUP_CLEANUP_INTERVAL
|
|
>= self._health_callback_interval
|
|
):
|
|
await self._report_health_if_callback()
|
|
cycle = 0
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
logger.debug(
|
|
"[%s] Dedup cleanup error (non-fatal)", self.name
|
|
)
|
|
|
|
def _prune_dedup_map(self) -> None:
|
|
"""Remove entries older than DEDUP_WINDOW from the dedup map."""
|
|
now = time.time()
|
|
cutoff = now - DEDUP_WINDOW
|
|
before = len(self._seen_message_ids)
|
|
self._seen_message_ids = {
|
|
k: v for k, v in self._seen_message_ids.items() if v > cutoff
|
|
}
|
|
after = len(self._seen_message_ids)
|
|
if before != after:
|
|
logger.debug(
|
|
"[%s] Dedup map pruned: %d -> %d entries",
|
|
self.name, before, after,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 3: Periodic @all-bots refresh
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _all_bots_refresh_forever(self) -> None:
|
|
"""Periodically re-resolve the @all-bots user ID.
|
|
|
|
Runs every ALL_BOTS_REFRESH_INTERVAL (1 hour by default).
|
|
Handles the case where the @all-bots user is created or
|
|
recreated after initial connection.
|
|
"""
|
|
while self._running:
|
|
try:
|
|
await asyncio.sleep(ALL_BOTS_REFRESH_INTERVAL)
|
|
await self._resolve_all_bots_user_id()
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception:
|
|
logger.debug(
|
|
"[%s] @all-bots refresh error (non-fatal)", self.name
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 5: Bot user ID resolution
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _resolve_bot_user_id(self) -> None:
|
|
"""Fetch the bot's own user ID from /api/v1/users/me.
|
|
|
|
The /register endpoint does not always return user_id on
|
|
all Zulip server versions. This ensures @mention detection
|
|
works by resolving it from the identity endpoint.
|
|
"""
|
|
try:
|
|
resp, _ = await self._api_call("GET", "/api/v1/users/me")
|
|
if resp:
|
|
uid = resp.get("user_id")
|
|
if uid is not None:
|
|
self._bot_user_id = int(uid)
|
|
logger.info(
|
|
"[%s] Resolved bot user_id=%s from /users/me",
|
|
self.name, uid,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(
|
|
"[%s] Could not resolve bot user_id: %s", self.name, e
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 2: Dynamic @all-bots resolution
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _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.
|
|
|
|
Falls back to the configured or default value if resolution fails.
|
|
"""
|
|
try:
|
|
resp, _ = await self._api_call("GET", "/api/v1/users")
|
|
if not resp:
|
|
return
|
|
members = resp.get("members", [])
|
|
for member in members:
|
|
email = member.get("email", "")
|
|
# @all-bots typically has an email pattern like all-bots@...
|
|
if "all-bots" in email.lower():
|
|
uid = member.get("user_id")
|
|
if uid is not None:
|
|
self._all_bots_user_id = int(uid)
|
|
self._health_stats["all_bots_refreshes"] += 1
|
|
logger.info(
|
|
"[%s] Resolved @all-bots user_id=%s from %s",
|
|
self.name, uid, email,
|
|
)
|
|
return
|
|
except Exception as e:
|
|
logger.debug(
|
|
"[%s] Could not resolve @all-bots ID dynamically: %s. "
|
|
"Using configured default %s.",
|
|
self.name, e, self._all_bots_user_id,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Message processing — Zulip event -> MessageEvent -> Gateway
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _process_zulip_event(self, event: Dict[str, Any]) -> None:
|
|
"""Convert a Zulip message event to MessageEvent and dispatch to gateway."""
|
|
msg = event.get("message", {})
|
|
msg_type = msg.get("type", "") # "private" or "stream"
|
|
sender_email = msg.get("sender_email", "")
|
|
sender_name = msg.get("sender_full_name", "Unknown")
|
|
sender_id = msg.get("sender_id")
|
|
content = msg.get("content", "")
|
|
|
|
# Echo-loop prevention: skip own messages
|
|
if sender_email == self._bot_email:
|
|
return
|
|
|
|
# Gen 2: Check for self-test response
|
|
if self._selftest_future and not self._selftest_future.done():
|
|
# Detect if this is a response to our self-test
|
|
self._selftest_future.set_result(True)
|
|
return
|
|
|
|
# Deduplication (O(1) — no cleanup on this path)
|
|
msg_id = f"{event.get('id', '')}:{msg.get('id', '')}"
|
|
if self._is_duplicate(msg_id):
|
|
return
|
|
|
|
# Update health stats
|
|
self._health_stats["messages_received"] += 1
|
|
self._health_stats["last_message_at"] = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Determine if this message targets this bot
|
|
mentioned_users = msg.get("mentioned_users", []) or []
|
|
mentioned_ids = [u.get("user_id") for u in mentioned_users if isinstance(u, dict)]
|
|
is_dm = msg_type == "private"
|
|
is_direct_mention = self._bot_user_id and self._bot_user_id in mentioned_ids
|
|
is_all_bots = self._all_bots_user_id in mentioned_ids
|
|
|
|
# DM-first: process all private messages
|
|
if is_dm:
|
|
self._health_stats["dms_routed"] += 1
|
|
logger.info("[%s] DM from %s: %.60s", self.name, sender_name, content)
|
|
await self._route_message(
|
|
text=content,
|
|
chat_id=str(sender_id),
|
|
chat_name=sender_name,
|
|
user_id=str(sender_id),
|
|
user_name=sender_name,
|
|
chat_type="dm",
|
|
msg_id=msg_id,
|
|
timestamp=msg.get("timestamp"),
|
|
raw=event,
|
|
)
|
|
return
|
|
|
|
# Stream: only respond to @mentions and @all-bots
|
|
if msg_type == "stream" and (is_direct_mention or is_all_bots):
|
|
self._health_stats["mentions_routed"] += 1
|
|
stream_name = msg.get("display_recipient", "unknown")
|
|
topic = msg.get("subject", "general")
|
|
logger.info(
|
|
"[%s] %s in #%s > %s: %.60s",
|
|
self.name,
|
|
"@mention" if is_direct_mention else "@all-bots",
|
|
stream_name, topic, content,
|
|
)
|
|
|
|
# Clean @mention artifacts from content
|
|
clean_content = MENTION_CLEANER.sub("", content).strip()
|
|
|
|
chat_id = f"{stream_name}:{topic}"
|
|
await self._route_message(
|
|
text=clean_content,
|
|
chat_id=chat_id,
|
|
chat_name=f"#{stream_name} > {topic}",
|
|
user_id=str(sender_id),
|
|
user_name=sender_name,
|
|
chat_type="stream",
|
|
thread_id=topic,
|
|
chat_topic=topic,
|
|
msg_id=msg_id,
|
|
timestamp=msg.get("timestamp"),
|
|
raw=event,
|
|
reply_to_id=msg.get("id"),
|
|
reply_to_type="stream",
|
|
reply_to_stream=stream_name,
|
|
reply_to_topic=topic,
|
|
)
|
|
|
|
async def _route_message(
|
|
self,
|
|
text: str,
|
|
chat_id: str,
|
|
chat_name: str,
|
|
user_id: str,
|
|
user_name: str,
|
|
chat_type: str,
|
|
msg_id: str,
|
|
timestamp: Any,
|
|
raw: Dict[str, Any],
|
|
thread_id: Optional[str] = None,
|
|
chat_topic: Optional[str] = None,
|
|
reply_to_id: Optional[int] = None,
|
|
reply_to_type: Optional[str] = None,
|
|
reply_to_stream: Optional[str] = None,
|
|
reply_to_topic: Optional[str] = None,
|
|
) -> None:
|
|
"""Build MessageEvent and send to gateway via handle_message."""
|
|
source = self.build_source(
|
|
chat_id=chat_id,
|
|
chat_name=chat_name,
|
|
chat_type=chat_type,
|
|
user_id=user_id,
|
|
user_name=user_name,
|
|
thread_id=thread_id,
|
|
chat_topic=chat_topic,
|
|
chat_id_alt=str(reply_to_id) if reply_to_id else None,
|
|
)
|
|
|
|
# Derive message type
|
|
message_type = MessageType.TEXT
|
|
|
|
message_event = MessageEvent(
|
|
text=text,
|
|
message_type=message_type,
|
|
source=source,
|
|
message_id=msg_id,
|
|
raw_message=raw,
|
|
timestamp=_parse_zulip_timestamp(timestamp),
|
|
)
|
|
|
|
# Store reply metadata so send() can respond to the right thread
|
|
if reply_to_id:
|
|
self._pending_replies[chat_id] = reply_to_id
|
|
|
|
logger.debug("[%s] Dispatching message from %s", self.name, user_name)
|
|
await self.handle_message(message_event)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Outbound messaging — send responses to Zulip
|
|
# ------------------------------------------------------------------
|
|
|
|
async def send(
|
|
self,
|
|
chat_id: str,
|
|
content: str,
|
|
reply_to: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
**kwargs: Any,
|
|
) -> SendResult:
|
|
"""Send a message to a Zulip chat.
|
|
|
|
For DMs, chat_id is the user_id. For streams, format is
|
|
"stream_name:topic". Sends a placeholder first if this is the
|
|
first message in a turn, then edits it on subsequent calls.
|
|
|
|
Accepts **kwargs for Hermes Gateway compatibility.
|
|
"""
|
|
# Determine target type and IDs
|
|
is_dm = not chat_id or ":" not in chat_id
|
|
truncated = _truncate(content)
|
|
|
|
if is_dm:
|
|
# DM: chat_id is the user_id
|
|
to = [int(chat_id)]
|
|
payload = {
|
|
"type": "private",
|
|
"to": json.dumps(to),
|
|
"content": truncated,
|
|
}
|
|
else:
|
|
# Stream: chat_id is "stream_name:topic"
|
|
parts = chat_id.split(":", 1)
|
|
stream_name = parts[0]
|
|
topic = parts[1] if len(parts) > 1 else "general"
|
|
payload = {
|
|
"type": "stream",
|
|
"to": stream_name,
|
|
"subject": topic,
|
|
"content": truncated,
|
|
}
|
|
|
|
# Check if we should send a placeholder first (first message in turn)
|
|
placeholder_msg_id = self._pending_replies.pop(chat_id, None)
|
|
|
|
if placeholder_msg_id:
|
|
# This is the first response — send placeholder and return its ID
|
|
placeholder = self._pick_placeholder()
|
|
placehold_payload = {**payload, "content": placeholder}
|
|
result = await self._send_api_call(placehold_payload)
|
|
if result and result.get("id"):
|
|
msg_id = str(result["id"])
|
|
logger.debug(
|
|
"[%s] Sent placeholder msg=%s for %s",
|
|
self.name, msg_id, chat_id,
|
|
)
|
|
self._health_stats["send_count"] += 1
|
|
return SendResult(
|
|
success=True,
|
|
message_id=msg_id,
|
|
)
|
|
|
|
# Send actual content
|
|
result = await self._send_api_call(payload)
|
|
if result and result.get("id"):
|
|
msg_id = str(result["id"])
|
|
self._health_stats["send_count"] += 1
|
|
return SendResult(
|
|
success=True,
|
|
message_id=msg_id,
|
|
)
|
|
|
|
self._health_stats["send_errors"] += 1
|
|
return SendResult(
|
|
success=False,
|
|
error="Failed to send message",
|
|
)
|
|
|
|
async def edit_message(
|
|
self,
|
|
chat_id: str,
|
|
message_id: str,
|
|
content: str,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
**kwargs: Any,
|
|
) -> SendResult:
|
|
"""Edit a previously sent Zulip message (for placeholder replacement).
|
|
|
|
Returns SendResult for Hermes Gateway streaming interface compatibility.
|
|
Accepts **kwargs (e.g., finalize=True) — extra kwargs are silently ignored.
|
|
|
|
Note: Zulip uses POST /api/v1/messages/{message_id} for editing.
|
|
"""
|
|
truncated = _truncate(content)
|
|
payload = {
|
|
"content": truncated,
|
|
}
|
|
try:
|
|
# Zulip API: PATCH /api/v1/messages/{message_id} with content in body
|
|
resp, status = await self._api_call(
|
|
"PATCH", f"/api/v1/messages/{int(message_id)}",
|
|
data=payload,
|
|
)
|
|
if resp:
|
|
self._health_stats["send_count"] += 1
|
|
return SendResult(
|
|
success=True,
|
|
message_id=str(resp.get("id", message_id)),
|
|
)
|
|
self._health_stats["send_errors"] += 1
|
|
return SendResult(
|
|
success=False,
|
|
error=f"Edit failed (status={status})",
|
|
)
|
|
except Exception as e:
|
|
logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e)
|
|
self._health_stats["send_errors"] += 1
|
|
return SendResult(
|
|
success=False,
|
|
error=str(e),
|
|
)
|
|
|
|
async def send_typing(self, chat_id: str, metadata: Optional[Dict] = None) -> None:
|
|
"""Send typing indicator to a Zulip chat."""
|
|
if ":" in chat_id:
|
|
return # No typing for stream messages
|
|
try:
|
|
user_ids = json.dumps([int(chat_id)])
|
|
await self._api_call(
|
|
"POST", "/api/v1/typing",
|
|
data={"to": user_ids, "op": "start"},
|
|
)
|
|
# typing indicator failure is non-critical
|
|
except Exception:
|
|
pass # Non-critical
|
|
|
|
async def stop_typing(self, chat_id: str) -> None:
|
|
"""Stop typing indicator."""
|
|
if ":" in chat_id:
|
|
return
|
|
try:
|
|
user_ids = json.dumps([int(chat_id)])
|
|
await self._api_call(
|
|
"POST", "/api/v1/typing",
|
|
data={"to": user_ids, "op": "stop"},
|
|
)
|
|
# typing indicator failure is non-critical
|
|
except Exception:
|
|
pass # Non-critical
|
|
|
|
async def get_chat_info(self, chat_id: str) -> dict:
|
|
"""Return basic info about a Zulip chat."""
|
|
if ":" in chat_id:
|
|
parts = chat_id.split(":", 1)
|
|
return {"name": parts[0], "type": "channel", "topic": parts[1]}
|
|
return {"name": chat_id, "type": "dm"}
|
|
|
|
async def delete_message(
|
|
self, chat_id: str, message_id: str,
|
|
metadata: Optional[Dict] = None,
|
|
**kwargs: Any,
|
|
) -> SendResult:
|
|
"""Delete a message via Zulip API."""
|
|
# Zulip doesn't support message deletion via API for bots
|
|
# Send an empty edit instead
|
|
return await self.edit_message(chat_id, message_id, "*deleted*", **kwargs)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 2: Self-test diagnostics
|
|
# ------------------------------------------------------------------
|
|
|
|
async def selftest(self) -> Dict[str, Any]:
|
|
"""Run a self-test to verify adapter health.
|
|
|
|
Checks:
|
|
1. Connection state (queue registered, HTTP client alive)
|
|
2. Bot identity resolved (bot_user_id is set)
|
|
3. Queue polling active (poll_task is running)
|
|
4. Echo-loop prevention configured (bot_email matches registered email)
|
|
5. Subscription to primary stream (if reachable)
|
|
6. @all-bots ID resolved
|
|
7. Dedup maintenance task running
|
|
|
|
Returns a dict with pass/fail for each check and an overall verdict.
|
|
"""
|
|
checks = {}
|
|
|
|
# 1. Connection state
|
|
checks["connected"] = {
|
|
"status": self._connected,
|
|
"detail": "Queue registered and connected"
|
|
if self._connected
|
|
else "Not connected to Zulip",
|
|
}
|
|
|
|
# 2. Queue registered
|
|
checks["queue_registered"] = {
|
|
"status": self._queue_id is not None,
|
|
"detail": f"Queue: {self._queue_id}"
|
|
if self._queue_id
|
|
else "No queue registered",
|
|
}
|
|
|
|
# 3. HTTP client alive
|
|
checks["http_client"] = {
|
|
"status": self._http_client is not None and not self._http_client.is_closed,
|
|
"detail": "HTTP client active"
|
|
if self._http_client and not self._http_client.is_closed
|
|
else "HTTP client unavailable",
|
|
}
|
|
|
|
# 4. Bot identity
|
|
checks["bot_identity"] = {
|
|
"status": self._bot_user_id is not None,
|
|
"detail": f"Bot user_id: {self._bot_user_id}"
|
|
if self._bot_user_id
|
|
else "Bot user_id not resolved",
|
|
}
|
|
|
|
# 5. Poll loop active
|
|
checks["poll_loop"] = {
|
|
"status": self._poll_task is not None and not self._poll_task.done(),
|
|
"detail": "Poll loop running"
|
|
if self._poll_task and not self._poll_task.done()
|
|
else "Poll loop not active",
|
|
}
|
|
|
|
# 6. Dedup cleanup active
|
|
checks["dedup_cleanup"] = {
|
|
"status": self._dedup_cleanup_task is not None
|
|
and not self._dedup_cleanup_task.done(),
|
|
"detail": "Dedup maintenance running"
|
|
if self._dedup_cleanup_task and not self._dedup_cleanup_task.done()
|
|
else "Dedup maintenance not active",
|
|
}
|
|
|
|
# 7. Echo-loop prevention
|
|
checks["echo_prevention"] = {
|
|
"status": bool(self._bot_email),
|
|
"detail": f"Bot email: {self._bot_email}",
|
|
}
|
|
|
|
# 8. @all-bots configured
|
|
checks["all_bots_configured"] = {
|
|
"status": self._all_bots_user_id is not None,
|
|
"detail": f"@all-bots user_id: {self._all_bots_user_id}",
|
|
}
|
|
|
|
# 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"])
|
|
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
|
|
- Known issues (Gen 6)
|
|
- Timestamps of last activity
|
|
|
|
Suitable for periodic logging to RA-H OS knowledge graph.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
|
|
uptime = None
|
|
if self._health_stats.get("started_at"):
|
|
try:
|
|
started = datetime.fromisoformat(self._health_stats["started_at"])
|
|
uptime_seconds = (now - started).total_seconds()
|
|
uptime = f"{uptime_seconds:.0f}s"
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
stats = dict(self._health_stats)
|
|
stats["uptime"] = uptime
|
|
stats["dedup_map_size"] = len(self._seen_message_ids)
|
|
stats["queue_id"] = self._queue_id
|
|
stats["bot_user_id"] = self._bot_user_id
|
|
stats["all_bots_user_id"] = self._all_bots_user_id
|
|
stats["connected"] = self._connected
|
|
stats["checked_at"] = now.isoformat()
|
|
stats["total_polls"] = self._total_poll_count
|
|
stats["consecutive_empty_polls"] = self._consecutive_empty_polls
|
|
stats["client_pool_resets"] = self._client_pool_reset_count
|
|
stats["silence_seconds"] = round(time.time() - self._last_event_received_time)
|
|
stats["known_issues"] = list(self._known_issues) # Gen 6
|
|
|
|
# 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
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 6: known_issues persistence
|
|
# ------------------------------------------------------------------
|
|
|
|
def add_known_issue(self, issue: str) -> None:
|
|
"""Record a discovered issue for tracking across generations.
|
|
|
|
Known issues persist in the adapter instance's lifetime and are
|
|
included in health stats for external logging to knowledge graph.
|
|
"""
|
|
if issue not in self._known_issues:
|
|
self._known_issues.append(issue)
|
|
logger.warning("[%s] Known issue recorded: %s", self.name, issue)
|
|
|
|
def get_known_issues(self) -> List[str]:
|
|
"""Return all tracked known issues."""
|
|
return list(self._known_issues)
|
|
|
|
def clear_known_issues(self) -> None:
|
|
"""Clear resolved issues from the tracker."""
|
|
if self._known_issues:
|
|
logger.info(
|
|
"[%s] Cleared %d known issues",
|
|
self.name, len(self._known_issues),
|
|
)
|
|
self._known_issues.clear()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Gen 6: Simulated DM latency test (offline-verifiable)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def simulate_dm_latency(self) -> Dict[str, Any]:
|
|
"""Simulate DM processing pipeline and measure round-trip timing.
|
|
|
|
Creates a mock DM event and routes it through the full processing
|
|
pipeline (parse → route → send placeholder) without a live Zulip
|
|
connection. Measures:
|
|
- Processing time (event parse → handle_message)
|
|
- Send pipeline time (placeholder generation)
|
|
- Total simulated round-trip
|
|
|
|
Returns timing breakdown for offline verification of the
|
|
dm_response_time_ms < 10000 postcondition.
|
|
"""
|
|
import time as _time
|
|
|
|
result = {"test": "simulated_dm_latency", "steps": {}}
|
|
|
|
# Step 1: Simulate message parse
|
|
t0 = _time.perf_counter()
|
|
mock_event = {
|
|
"id": 99999,
|
|
"type": "message",
|
|
"message": {
|
|
"id": 88888,
|
|
"type": "private",
|
|
"sender_email": "test@sysloggh.net",
|
|
"sender_full_name": "Test User",
|
|
"sender_id": 42,
|
|
"content": "Hello, this is a test message for latency measurement.",
|
|
"timestamp": _time.time(),
|
|
"mentioned_users": [],
|
|
},
|
|
}
|
|
t1 = _time.perf_counter()
|
|
result["steps"]["event_parse"] = round((t1 - t0) * 1000, 2)
|
|
|
|
# Step 2: Simulate placeholder pick
|
|
t0 = _time.perf_counter()
|
|
placeholder = self._pick_placeholder()
|
|
t1 = _time.perf_counter()
|
|
result["steps"]["placeholder_pick"] = round((t1 - t0) * 1000, 2)
|
|
result["placeholder"] = placeholder
|
|
|
|
# Step 3: Simulate auth header (typically done once, measure for completeness)
|
|
t0 = _time.perf_counter()
|
|
if self._email and self._api_key:
|
|
_build_auth_header(self._email, self._api_key)
|
|
t1 = _time.perf_counter()
|
|
result["steps"]["auth_header"] = round((t1 - t0) * 1000, 2)
|
|
|
|
# Step 4: Simulate dedup check
|
|
t0 = _time.perf_counter()
|
|
is_dup = self._is_duplicate("99999:88888")
|
|
t1 = _time.perf_counter()
|
|
result["steps"]["dedup_check"] = round((t1 - t0) * 1000, 2)
|
|
result["is_duplicate"] = is_dup
|
|
|
|
# Step 5: Simulate truncation (worst case)
|
|
t0 = _time.perf_counter()
|
|
_truncate("x" * 15000)
|
|
t1 = _time.perf_counter()
|
|
result["steps"]["truncation"] = round((t1 - t0) * 1000, 2)
|
|
|
|
# Aggregate
|
|
total_ms = sum(result["steps"].values())
|
|
result["total_simulated_ms"] = round(total_ms, 2)
|
|
result["dm_response_time_ms_pass"] = total_ms < 10000
|
|
|
|
logger.info(
|
|
"[%s] Simulated DM latency: %.2fms (%s)",
|
|
self.name, total_ms,
|
|
"PASS" if result["dm_response_time_ms_pass"] else "FAIL",
|
|
)
|
|
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Zulip API helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _api_call(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
data: Optional[Dict] = None,
|
|
params: Optional[Dict] = None,
|
|
) -> Tuple[Optional[Dict], int]:
|
|
"""Make an API call to Zulip.
|
|
|
|
Returns (response_json, status_code). status_code is useful for
|
|
callers that need to detect specific HTTP errors like 400
|
|
(BAD_EVENT_QUEUE_ID). Returns (None, 0) on transport errors.
|
|
"""
|
|
if not self._http_client:
|
|
return None, 0
|
|
|
|
url = f"{self._site}{path}"
|
|
headers = {
|
|
"Authorization": self._auth_header,
|
|
}
|
|
|
|
try:
|
|
if method == "GET":
|
|
response = await self._http_client.get(
|
|
url, params=params, headers=headers,
|
|
)
|
|
elif method == "POST":
|
|
response = await self._http_client.post(
|
|
url, data=data, headers=headers,
|
|
)
|
|
elif method == "PATCH":
|
|
response = await self._http_client.patch(
|
|
url, data=data, headers=headers,
|
|
)
|
|
else:
|
|
return None, 0
|
|
|
|
if response.status_code >= 400:
|
|
# Gen 4: Truncate error bodies to prevent HTML spam
|
|
body = response.text[:MAX_ERROR_LOG_LEN].replace("\n", " ")
|
|
logger.warning(
|
|
"[%s] API %s %s: %d %s",
|
|
self.name, method, path,
|
|
response.status_code, body,
|
|
)
|
|
return None, response.status_code
|
|
|
|
return response.json(), response.status_code
|
|
|
|
except httpx.TimeoutException:
|
|
logger.debug("[%s] Timeout on %s %s", self.name, method, path)
|
|
return None, 0
|
|
except Exception as e:
|
|
logger.warning("[%s] API error %s %s: %s", self.name, method, path, e)
|
|
return None, 0
|
|
|
|
async def _send_api_call(self, payload: Dict) -> Optional[Dict]:
|
|
"""Send a message to Zulip."""
|
|
result, _ = await self._api_call("POST", "/api/v1/messages", data=payload)
|
|
return result
|
|
|
|
# ------------------------------------------------------------------
|
|
# Utilities
|
|
# ------------------------------------------------------------------
|
|
|
|
def _pick_placeholder(self) -> str:
|
|
"""Pick a random placeholder message for streaming UX.
|
|
|
|
Gen 6: Switched from hash-based to random.choice() to avoid
|
|
collision when called twice in the same second.
|
|
"""
|
|
return random.choice(PLACEHOLDERS)
|
|
|
|
def _is_duplicate(self, msg_id: str) -> bool:
|
|
"""Deduplication using message IDs — O(1) lookup, no cleanup.
|
|
|
|
Cleanup runs in a separate background task (_dedup_cleanup_forever)
|
|
so this path stays fast.
|
|
"""
|
|
if msg_id in self._seen_message_ids:
|
|
return True
|
|
self._seen_message_ids[msg_id] = time.time()
|
|
return False
|
|
|
|
def get_all_bots_user_id(self) -> int:
|
|
"""Return the current @all-bots user ID.
|
|
|
|
This is used by the zulip-mention-reliability contract for
|
|
cross-contract verification.
|
|
"""
|
|
return self._all_bots_user_id
|
|
|
|
def get_bot_user_id(self) -> Optional[int]:
|
|
"""Return the bot's own user ID for identity checks."""
|
|
return self._bot_user_id
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Plugin registration helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def check_requirements() -> bool:
|
|
"""Check whether the Zulip adapter is installable."""
|
|
if not HTTPX_AVAILABLE:
|
|
return False
|
|
site = os.getenv("ZULIP_SITE", "").strip() or os.getenv("ZULIP_URL", "").strip()
|
|
email = os.getenv("ZULIP_EMAIL", "").strip()
|
|
api_key = os.getenv("ZULIP_API_KEY", "").strip()
|
|
return bool(site and email and api_key)
|
|
|
|
|
|
def validate_config(config) -> bool:
|
|
"""Validate that the configured Zulip platform has credentials."""
|
|
extra = getattr(config, "extra", {}) or {}
|
|
site = extra.get("site") or os.getenv("ZULIP_SITE", "") or os.getenv("ZULIP_URL", "")
|
|
email = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
|
|
api_key = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")
|
|
return bool(site and email and api_key)
|
|
|
|
|
|
def is_connected(config) -> bool:
|
|
"""Check whether Zulip is configured (env or config.yaml)."""
|
|
return validate_config(config)
|
|
|
|
|
|
def _env_enablement() -> Optional[dict]:
|
|
"""Seeds PlatformConfig.extra from env vars for env-only setups."""
|
|
site = os.getenv("ZULIP_SITE", "").strip() or os.getenv("ZULIP_URL", "").strip()
|
|
email = os.getenv("ZULIP_EMAIL", "").strip()
|
|
api_key = os.getenv("ZULIP_API_KEY", "").strip()
|
|
if not (site and email and api_key):
|
|
return None
|
|
result: Dict[str, Any] = {
|
|
"site": site,
|
|
"email": email,
|
|
"api_key": api_key,
|
|
}
|
|
stream = os.getenv("ZULIP_STREAM", "").strip()
|
|
if stream:
|
|
result["stream"] = stream
|
|
agent_name = os.getenv("ZULIP_AGENT_NAME", "").strip()
|
|
if agent_name:
|
|
result["agent_name"] = agent_name
|
|
home = os.getenv("ZULIP_HOME_CHANNEL", "").strip()
|
|
if home:
|
|
result["home_channel"] = home
|
|
return result
|
|
|
|
|
|
def register(ctx) -> None:
|
|
"""Plugin entry point — called by the Hermes plugin system at startup."""
|
|
ctx.register_platform(
|
|
name="zulip",
|
|
label="Zulip",
|
|
adapter_factory=lambda cfg: ZulipAdapter(cfg),
|
|
check_fn=check_requirements,
|
|
validate_config=validate_config,
|
|
is_connected=is_connected,
|
|
required_env=["ZULIP_SITE", "ZULIP_EMAIL", "ZULIP_API_KEY"],
|
|
install_hint="pip install httpx # already a Hermes dependency",
|
|
env_enablement_fn=_env_enablement,
|
|
cron_deliver_env_var="ZULIP_HOME_CHANNEL",
|
|
allowed_users_env="ZULIP_ALLOWED_USERS",
|
|
allow_all_env="ZULIP_ALLOW_ALL_USERS",
|
|
max_message_length=MAX_ZULIP_MESSAGE,
|
|
emoji="💬",
|
|
pii_safe=True,
|
|
allow_update_command=True,
|
|
platform_hint=(
|
|
"You are communicating via Zulip. Format responses with "
|
|
"Zulip-compatible Markdown. Keep responses concise and "
|
|
"well-structured. DMs are private conversations; stream "
|
|
"messages are visible to all subscribers."
|
|
),
|
|
)
|