zulip-platform-plugins/plugins/platforms/zulip/adapter.py

"""
Zulip platform adapter (Hermes plugin) — Gen 4.

Connects a Hermes agent to Zulip via event queue polling. DMs and @mentions
are routed to the agent via the Hermes Gateway's handle_message() interface.
Replies use a placeholder->edit streaming pattern for UX feedback.

Gen 4 Improvements (2026-06-27):
  1. HTTP client recreation on reconnect — new httpx.AsyncClient() after queue expiry
     or sustained failures to prevent CLOSE-WAIT socket leaks
  2. Heartbeat logging — periodic "still alive" message to prove poll loop is running
  3. Log sanitization — truncates 502/error HTML to first 80 chars (no Netbird HTML spam)
  4. Connection recovery — creates fresh HTTP client after N empty poll cycles
  5. Poll silence detection — logs warnings when no events received for extended period
  6. Connection reuse limit — periodic client recreation to prevent connection pool exhaustion

Gen 3 Improvements (2026-06-26):
  1. Malformed message resilience — try/except wraps each event, no crash on bad JSON
  2. Periodic @all-bots refresh — background task re-resolves user ID every hour
  3. Health stats callback — external consumers can register for periodic stats

Gen 2 Improvements (2026-06-26):
  1. Background dedup maintenance — O(1) per-message path, periodic cleanup task
  2. Self-test capability — selftest() method verifying DM/mention/queue health
  3. Health stats tracking — poll stats, reconnect counts, error rates
  4. Dynamic @all-bots resolution — falls back to hardcoded default

Architecture (mirrors pi-zulip-extension):
  Zulip event queue -> poll loop -> parse message -> handle_message(MessageEvent)
  -> Gateway processes -> send() / edit_message() posts response to Zulip

No external SDK required — only httpx (already a Hermes dependency).
Ships as a Hermes platform plugin under plugins/platforms/zulip/.
"""

import asyncio
import base64
import json
import logging
import os
import re
import time
import uuid
from collections import deque
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

try:
    import httpx
    HTTPX_AVAILABLE = True
except ImportError:
    HTTPX_AVAILABLE = False
    httpx = None

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter,
    MessageEvent,
    MessageType,
    SendResult,
)

logger = logging.getLogger(__name__)

# Constants
DEFAULT_STREAM = "agent-hub"
DEFAULT_ALL_BOTS_USER_ID = 1
DEFAULT_POLL_INTERVAL = 3.0
MAX_ZULIP_MESSAGE = 10000
TRUNCATION_NOTICE = "\n\n[...truncated at Zulip limit]"
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 - len(TRUNCATION_NOTICE)] + TRUNCATION_NOTICE


def _parse_zulip_timestamp(ts: Any) -> datetime:
    """Convert Zulip timestamp (float seconds) to datetime."""
    try:
        return datetime.fromtimestamp(float(ts), tz=timezone.utc)
    except (ValueError, OSError, TypeError):
        return datetime.now(tz=timezone.utc)


class ZulipAdapter(BasePlatformAdapter):
    """Zulip adapter for Hermes agents.

    Connects to Zulip via event queues, polls for new events, converts
    DMs and @mentions to MessageEvent, and sends responses with streaming
    placeholder->edit UX.

    Gen 3 adds:
    - Malformed message resilience (per-event try/except)
    - Periodic @all-bots refresh (hourly background task)
    - Health stats callback for external consumers

    Gen 2 adds:
    - Background dedup maintenance task (O(1) per-message)
    - Self-test diagnostics (DM/mention/queue health)
    - Health stats for monitoring
    - Dynamic @all-bots resolution
    """

    def __init__(self, config: PlatformConfig):
        platform = Platform("zulip")
        super().__init__(config=config, platform=platform)

        extra = config.extra or {}

        # Zulip connection
        self._site: str = (
            extra.get("site") or os.getenv("ZULIP_SITE", "")
        ).rstrip("/")
        self._email: str = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
        self._api_key: str = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")

        # Agent identity
        self._agent_name: str = (
            extra.get("agent_name") or os.getenv("ZULIP_AGENT_NAME", "hermes-agent")
        )

        # Stream config
        self._stream: str = (
            extra.get("stream") or os.getenv("ZULIP_STREAM", DEFAULT_STREAM)
        )

        # Self-test config — owner user ID for DM self-tests
        self._owner_user_id: Optional[int] = self._resolve_int_opt(
            extra.get("owner_user_id") or os.getenv("ZULIP_OWNER_USER_ID", "")
        )

        # @all-bots user ID — dynamically resolved on connect, falls back to config/env/default
        self._all_bots_user_id: int = self._resolve_int(
            extra.get("all_bots_user_id")
            or os.getenv("ZULIP_ALL_BOTS_USER_ID", str(DEFAULT_ALL_BOTS_USER_ID)),
            DEFAULT_ALL_BOTS_USER_ID,
        )

        # Polling
        self._poll_interval: float = float(
            extra.get("poll_interval")
            or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL))
        )

        # --- Gen 4: Connection health ---
        self._consecutive_empty_polls: int = 0
        self._total_poll_count: int = 0
        self._last_event_received_time: float = time.time()
        self._last_heartbeat_time: float = 0.0
        self._client_pool_reset_count: int = 0

        # --- State ---
        # Support both ZULIP_SITE and ZULIP_URL env vars
        if not self._site:
            self._site = (os.getenv("ZULIP_URL", "") or "").rstrip("/")

        self._auth_header: str = _build_auth_header(self._email, self._api_key)
        self._queue_id: Optional[str] = None
        self._last_event_id: int = -1
        self._poll_task: Optional[asyncio.Task] = None
        self._dedup_cleanup_task: Optional[asyncio.Task] = None
        self._http_client: Optional[httpx.AsyncClient] = None
        self._pending_replies: Dict[str, int] = {}  # chat_id -> placeholder zulip_msg_id

        # Derived identity for echo-loop prevention
        self._bot_user_id: Optional[int] = None
        self._bot_email: str = self._email

        # --- Gen 2: Dedup (background-maintained) ---
        self._seen_message_ids: Dict[str, float] = {}

        # --- Gen 2/3: Health stats ---
        self._health_stats: Dict[str, Any] = {
            "started_at": None,
            "poll_count": 0,
            "poll_errors": 0,
            "malformed_events": 0,
            "messages_received": 0,
            "dms_routed": 0,
            "mentions_routed": 0,
            "reconnects": 0,
            "send_count": 0,
            "send_errors": 0,
            "all_bots_refreshes": 0,
            "last_poll_at": None,
            "last_message_at": None,
            "last_error_at": None,
            "last_error_msg": None,
        }

        # --- Gen 2: Self-test future for async response ---
        self._selftest_future: Optional[asyncio.Future] = None

        # --- Gen 3: Periodic @all-bots refresh ---
        self._all_bots_refresh_task: Optional[asyncio.Task] = None

        # --- Gen 3: Health stats callback ---
        self._health_callback = None
        self._health_callback_interval: int = 300  # 5 min

    @staticmethod
    def _resolve_int(val: Any, default: int) -> int:
        """Resolve an integer config value, falling back to default."""
        if val is None or val == "":
            return default
        try:
            return int(val)
        except (ValueError, TypeError):
            return default

    @staticmethod
    def _resolve_int_opt(val: Any) -> Optional[int]:
        """Resolve an optional integer config value."""
        if val is None or val == "":
            return None
        try:
            return int(val)
        except (ValueError, TypeError):
            return None

    # ------------------------------------------------------------------
    # Connection lifecycle
    # ------------------------------------------------------------------

    async def connect(self, **kwargs: Any) -> bool:
        """Connect to Zulip and register an event queue.

        Accepts **kwargs for Hermes Gateway compatibility (e.g., is_reconnect).
        """
        if not HTTPX_AVAILABLE:
            logger.warning("[%s] httpx not installed", self.name)
            return False

        if not self._site or not self._email or not self._api_key:
            logger.warning("[%s] Zulip credentials not configured", self.name)
            return False

        try:
            await self._create_http_client()
            self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat()

            # Reset connection health counters
            self._consecutive_empty_polls = 0
            self._total_poll_count = 0
            self._last_event_received_time = time.time()

            # Register event queue
            queue_resp, _ = await self._api_call(
                "POST", "/api/v1/register",
                data={
                    "event_types": '["message"]',
                    "apply_markdown": "true",
                    "include_subscribers": "false",
                },
            )
            if not queue_resp:
                logger.error("[%s] Failed to register event queue", self.name)
                return False

            data = queue_resp
            self._queue_id = data.get("queue_id")
            self._last_event_id = data.get("last_event_id", -1)
            self._bot_user_id = data.get("user_id")

            # Resolve bot user_id if register endpoint didn't provide it
            # (common on some Zulip versions)
            if self._bot_user_id is None:
                await self._resolve_bot_user_id()

            # Gen 2: Try to resolve @all-bots user ID dynamically
            await self._resolve_all_bots_user_id()

            self._mark_connected()
            logger.info(
                "[%s] Connected to %s as %s (queue=%s, bot_id=%s, all_bots_id=%s)",
                self.name, self._site, self._email,
                self._queue_id, self._bot_user_id, self._all_bots_user_id,
            )

            # Start poll loop
            self._poll_task = asyncio.create_task(self._poll_forever())

            # Gen 2: Start background dedup cleanup task
            self._dedup_cleanup_task = asyncio.create_task(
                self._dedup_cleanup_forever()
            )

            # Gen 3: Start periodic @all-bots refresh task
            self._all_bots_refresh_task = asyncio.create_task(
                self._all_bots_refresh_forever()
            )

            return True

        except Exception as e:
            logger.error("[%s] Connection failed: %s", self.name, e)
            return False

    async def disconnect(self) -> None:
        """Disconnect from Zulip and cancel all background tasks."""
        # Cancel dedup cleanup task
        if self._dedup_cleanup_task:
            self._dedup_cleanup_task.cancel()
            try:
                await self._dedup_cleanup_task
            except asyncio.CancelledError:
                pass
            self._dedup_cleanup_task = None

        # Cancel @all-bots refresh task
        if self._all_bots_refresh_task:
            self._all_bots_refresh_task.cancel()
            try:
                await self._all_bots_refresh_task
            except asyncio.CancelledError:
                pass
            self._all_bots_refresh_task = None

        # Cancel poll task
        if self._poll_task:
            self._poll_task.cancel()
            try:
                await self._poll_task
            except asyncio.CancelledError:
                pass
            self._poll_task = None

        self._queue_id = None
        self._mark_disconnected()
        logger.info("[%s] Disconnected", self.name)

    # ------------------------------------------------------------------
    # Poll loop — receives messages from Zulip
    # ------------------------------------------------------------------

    async def _poll_forever(self) -> None:
        """Continuously poll the Zulip event queue."""
        backoff_idx = 0

        while self._running:
            if not self._queue_id:
                await asyncio.sleep(self._poll_interval)
                continue

            try:
                events = await self._fetch_events()
                self._health_stats["poll_count"] += 1
                self._total_poll_count += 1
                self._health_stats["last_poll_at"] = datetime.now(
                    timezone.utc
                ).isoformat()

                if events:
                    now = time.time()
                    self._consecutive_empty_polls = 0
                    self._last_event_received_time = now
                    for event in events:
                        try:
                            await self._process_zulip_event(event)
                        except Exception as e:
                            # Gen 3: Malformed message resilience — catch per-event
                            # failures so one bad message never kills the poll loop
                            logger.warning(
                                "[%s] Malformed event skipped: %s. "
                                "Event id=%s",
                                self.name, e,
                                event.get("id", "unknown"),
                            )
                            self._health_stats["malformed_events"] = (
                                self._health_stats.get("malformed_events", 0) + 1
                            )
                else:
                    # Gen 4: Track consecutive empty polls to detect silent failures
                    self._consecutive_empty_polls += 1

                    # Gen 4: Detect prolonged silence (no events for too long)
                    silence_duration = time.time() - self._last_event_received_time
                    if silence_duration > POLL_SILENCE_WARN_INTERVAL:
                        logger.warning(
                            "[%s] No events received for %.0fs "
                            "(consecutive_empty=%d, total_polls=%d, "
                            "queue=%s)",
                            self.name, silence_duration,
                            self._consecutive_empty_polls,
                            self._total_poll_count,
                            self._queue_id,
                        )

                    # Gen 4: Reset HTTP client if too many empty polls in a row
                    # (connection pool likely stuck in CLOSE-WAIT)
                    if (
                        self._consecutive_empty_polls
                        >= MAX_CONSECUTIVE_EMPTY_POLLS
                    ):
                        logger.warning(
                            "[%s] %d consecutive empty polls — "
                            "recreating HTTP client and reconnecting",
                            self.name, self._consecutive_empty_polls,
                        )
                        await self._reconnect_with_fresh_client()
                        backoff_idx = 0
                        self._consecutive_empty_polls = 0
                        continue

                    # Gen 4: Periodic client refresh to prevent connection leaks
                    if self._total_poll_count % CLIENT_REUSE_LIMIT == 0:
                        logger.info(
                            "[%s] Client reuse limit reached (%d polls) — "
                            "creating fresh HTTP client",
                            self.name, self._total_poll_count,
                        )
                        old_client = self._http_client
                        await self._create_http_client()
                        if old_client:
                            try:
                                await old_client.aclose()
                            except Exception:
                                pass

                # Gen 4: Heartbeat logging — prove poll loop is alive
                now = time.time()
                if now - self._last_heartbeat_time > HEARTBEAT_INTERVAL:
                    self._last_heartbeat_time = now
                    silence = now - self._last_event_received_time
                    logger.info(
                        "[%s] Heartbeat — polls=%d empty=%d "
                        "silence=%.0fs errors=%d reconnects=%d "
                        "queue=%s",
                        self.name, self._total_poll_count,
                        self._consecutive_empty_polls, silence,
                        self._health_stats["poll_errors"],
                        self._health_stats["reconnects"],
                        self._queue_id,
                    )

                backoff_idx = 0
            except asyncio.CancelledError:
                break
            except Exception as e:
                if not self._running:
                    break
                self._health_stats["poll_errors"] += 1
                self._health_stats["last_error_at"] = datetime.now(
                    timezone.utc
                ).isoformat()
                self._health_stats["last_error_msg"] = str(e)[:200]

                err_str = str(e)
                if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower():
                    logger.info("[%s] Queue expired, re-registering...", self.name)
                    await self._reconnect_with_fresh_client()
                    backoff_idx = 0
                    continue
                logger.warning("[%s] Poll error: %s", self.name, e)
                delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF) - 1)]
                backoff_idx += 1
                await asyncio.sleep(delay)

            await asyncio.sleep(self._poll_interval)

    async def _fetch_events(self) -> List[Dict[str, Any]]:
        """Fetch events from the Zulip event queue.

        Raises RuntimeError with BAD_EVENT_QUEUE_ID when the queue
        expires, so _poll_forever can reconnect.
        """
        if not self._queue_id:
            return []

        resp, raw_status = await self._api_call(
            "GET", "/api/v1/events",
            params={
                "queue_id": self._queue_id,
                "last_event_id": str(self._last_event_id),
                "dont_block": "true",
            },
        )
        # Detect queue expiry from HTTP response
        if raw_status == 400:
            raise RuntimeError("BAD_EVENT_QUEUE_ID: queue expired")
        if not resp:
            return []

        events = resp.get("events", [])
        # Track last event ID for cursor advancement
        for event in events:
            eid = event.get("id", 0)
            if eid > self._last_event_id:
                self._last_event_id = eid

        return [e for e in events if e.get("type") == "message"]

    async def _create_http_client(self) -> None:
        """Create a fresh httpx client, closing the old one if it exists.

        Gen 4: This ensures a clean connection pool after sustained failures
        or queue expiry, preventing CLOSE-WAIT socket leaks.
        """
        old_client = self._http_client
        self._http_client = httpx.AsyncClient(
            timeout=httpx.Timeout(30.0, connect=15.0),
            limits=httpx.Limits(
                max_keepalive_connections=5,
                max_connections=10,
                keepalive_expiry=60.0,
            ),
        )
        self._client_pool_reset_count += 1
        if old_client:
            try:
                await old_client.aclose()
            except Exception:
                pass

    async def _reconnect(self) -> None:
        """Re-register the event queue."""
        self._queue_id = None
        try:
            resp, _ = await self._api_call(
                "POST", "/api/v1/register",
                data={
                    "event_types": '["message"]',
                    "apply_markdown": "true",
                    "include_subscribers": "false",
                },
            )
            if resp:
                self._queue_id = resp.get("queue_id")
                self._last_event_id = resp.get("last_event_id", -1)
                self._health_stats["reconnects"] += 1
                logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id)
                self._consecutive_empty_polls = 0
                self._last_event_received_time = time.time()
        except Exception as e:
            logger.error("[%s] Reconnect failed: %s", self.name, e)

    async def _reconnect_with_fresh_client(self) -> None:
        """Re-register the event queue with a fresh HTTP client.

        Gen 4: Creates a new httpx client to break out of stuck connection
        pools (CLOSE-WAIT sockets from previous failures).
        """
        logger.info("[%s] Reconnecting with fresh HTTP client", self.name)
        await self._create_http_client()
        await self._reconnect()

    # ------------------------------------------------------------------
    # Gen 2: Background dedup maintenance
    # ------------------------------------------------------------------

    async def _dedup_cleanup_forever(self) -> None:
        """Periodically purge stale entries from the dedup map.

        This runs as a background task so the per-message _is_duplicate()
        path stays O(1) — no cleanup cost on every message.

        Also triggers the health stats callback on each cycle if one is
        registered — natural tick for periodic monitoring.
        """
        cycle = 0
        while self._running:
            try:
                await asyncio.sleep(DEDUP_CLEANUP_INTERVAL)
                self._prune_dedup_map()
                cycle += 1
                # Report health every N cycles based on callback interval
                if (
                    self._health_callback
                    and cycle * DEDUP_CLEANUP_INTERVAL
                    >= self._health_callback_interval
                ):
                    await self._report_health_if_callback()
                    cycle = 0
            except asyncio.CancelledError:
                break
            except Exception:
                logger.debug(
                    "[%s] Dedup cleanup error (non-fatal)", self.name
                )

    def _prune_dedup_map(self) -> None:
        """Remove entries older than DEDUP_WINDOW from the dedup map."""
        now = time.time()
        cutoff = now - DEDUP_WINDOW
        before = len(self._seen_message_ids)
        self._seen_message_ids = {
            k: v for k, v in self._seen_message_ids.items() if v > cutoff
        }
        after = len(self._seen_message_ids)
        if before != after:
            logger.debug(
                "[%s] Dedup map pruned: %d -> %d entries",
                self.name, before, after,
            )

    # ------------------------------------------------------------------
    # Gen 3: Periodic @all-bots refresh
    # ------------------------------------------------------------------

    async def _all_bots_refresh_forever(self) -> None:
        """Periodically re-resolve the @all-bots user ID.

        Runs every ALL_BOTS_REFRESH_INTERVAL (1 hour by default).
        Handles the case where the @all-bots user is created or
        recreated after initial connection.
        """
        while self._running:
            try:
                await asyncio.sleep(ALL_BOTS_REFRESH_INTERVAL)
                await self._resolve_all_bots_user_id()
            except asyncio.CancelledError:
                break
            except Exception:
                logger.debug(
                    "[%s] @all-bots refresh error (non-fatal)", self.name
                )

    # ------------------------------------------------------------------
    # Gen 5: Bot user ID resolution
    # ------------------------------------------------------------------

    async def _resolve_bot_user_id(self) -> None:
        """Fetch the bot's own user ID from /api/v1/users/me.

        The /register endpoint does not always return user_id on
        all Zulip server versions. This ensures @mention detection
        works by resolving it from the identity endpoint.
        """
        try:
            resp, _ = await self._api_call("GET", "/api/v1/users/me")
            if resp:
                uid = resp.get("user_id")
                if uid is not None:
                    self._bot_user_id = int(uid)
                    logger.info(
                        "[%s] Resolved bot user_id=%s from /users/me",
                        self.name, uid,
                    )
        except Exception as e:
            logger.debug(
                "[%s] Could not resolve bot user_id: %s", self.name, e
            )

    # ------------------------------------------------------------------
    # Gen 2: Dynamic @all-bots resolution
    # ------------------------------------------------------------------

    async def _resolve_all_bots_user_id(self) -> None:
        """Try to resolve the @all-bots user ID from the Zulip server.

        Falls back to the configured or default value if resolution fails.
        """
        try:
            resp, _ = await self._api_call("GET", "/api/v1/users")
            if not resp:
                return
            members = resp.get("members", [])
            for member in members:
                email = member.get("email", "")
                # @all-bots typically has an email pattern like all-bots@...
                if "all-bots" in email.lower():
                    uid = member.get("user_id")
                    if uid is not None:
                        self._all_bots_user_id = int(uid)
                        self._health_stats["all_bots_refreshes"] += 1
                        logger.info(
                            "[%s] Resolved @all-bots user_id=%s from %s",
                            self.name, uid, email,
                        )
                        return
        except Exception as e:
            logger.debug(
                "[%s] Could not resolve @all-bots ID dynamically: %s. "
                "Using configured default %s.",
                self.name, e, self._all_bots_user_id,
            )

    # ------------------------------------------------------------------
    # Message processing — Zulip event -> MessageEvent -> Gateway
    # ------------------------------------------------------------------

    async def _process_zulip_event(self, event: Dict[str, Any]) -> None:
        """Convert a Zulip message event to MessageEvent and dispatch to gateway."""
        msg = event.get("message", {})
        msg_type = msg.get("type", "")  # "private" or "stream"
        sender_email = msg.get("sender_email", "")
        sender_name = msg.get("sender_full_name", "Unknown")
        sender_id = msg.get("sender_id")
        content = msg.get("content", "")

        # Echo-loop prevention: skip own messages
        if sender_email == self._bot_email:
            return

        # Gen 2: Check for self-test response
        if self._selftest_future and not self._selftest_future.done():
            # Detect if this is a response to our self-test
            self._selftest_future.set_result(True)
            return

        # Deduplication (O(1) — no cleanup on this path)
        msg_id = f"{event.get('id', '')}:{msg.get('id', '')}"
        if self._is_duplicate(msg_id):
            return

        # Update health stats
        self._health_stats["messages_received"] += 1
        self._health_stats["last_message_at"] = datetime.now(timezone.utc).isoformat()

        # Determine if this message targets this bot
        mentioned_users = msg.get("mentioned_users", []) or []
        mentioned_ids = [u.get("user_id") for u in mentioned_users if isinstance(u, dict)]
        is_dm = msg_type == "private"
        is_direct_mention = self._bot_user_id and self._bot_user_id in mentioned_ids
        is_all_bots = self._all_bots_user_id in mentioned_ids

        # DM-first: process all private messages
        if is_dm:
            self._health_stats["dms_routed"] += 1
            logger.info("[%s] DM from %s: %.60s", self.name, sender_name, content)
            await self._route_message(
                text=content,
                chat_id=str(sender_id),
                chat_name=sender_name,
                user_id=str(sender_id),
                user_name=sender_name,
                chat_type="dm",
                msg_id=msg_id,
                timestamp=msg.get("timestamp"),
                raw=event,
            )
            return

        # Stream: only respond to @mentions and @all-bots
        if msg_type == "stream" and (is_direct_mention or is_all_bots):
            self._health_stats["mentions_routed"] += 1
            stream_name = msg.get("display_recipient", "unknown")
            topic = msg.get("subject", "general")
            logger.info(
                "[%s] %s in #%s > %s: %.60s",
                self.name,
                "@mention" if is_direct_mention else "@all-bots",
                stream_name, topic, content,
            )

            # Clean @mention artifacts from content
            clean_content = MENTION_CLEANER.sub("", content).strip()

            chat_id = f"{stream_name}:{topic}"
            await self._route_message(
                text=clean_content,
                chat_id=chat_id,
                chat_name=f"#{stream_name} > {topic}",
                user_id=str(sender_id),
                user_name=sender_name,
                chat_type="stream",
                thread_id=topic,
                chat_topic=topic,
                msg_id=msg_id,
                timestamp=msg.get("timestamp"),
                raw=event,
                reply_to_id=msg.get("id"),
                reply_to_type="stream",
                reply_to_stream=stream_name,
                reply_to_topic=topic,
            )

    async def _route_message(
        self,
        text: str,
        chat_id: str,
        chat_name: str,
        user_id: str,
        user_name: str,
        chat_type: str,
        msg_id: str,
        timestamp: Any,
        raw: Dict[str, Any],
        thread_id: Optional[str] = None,
        chat_topic: Optional[str] = None,
        reply_to_id: Optional[int] = None,
        reply_to_type: Optional[str] = None,
        reply_to_stream: Optional[str] = None,
        reply_to_topic: Optional[str] = None,
    ) -> None:
        """Build MessageEvent and send to gateway via handle_message."""
        source = self.build_source(
            chat_id=chat_id,
            chat_name=chat_name,
            chat_type=chat_type,
            user_id=user_id,
            user_name=user_name,
            thread_id=thread_id,
            chat_topic=chat_topic,
            chat_id_alt=str(reply_to_id) if reply_to_id else None,
        )

        # Derive message type
        message_type = MessageType.TEXT

        message_event = MessageEvent(
            text=text,
            message_type=message_type,
            source=source,
            message_id=msg_id,
            raw_message=raw,
            timestamp=_parse_zulip_timestamp(timestamp),
        )

        # Store reply metadata so send() can respond to the right thread
        if reply_to_id:
            self._pending_replies[chat_id] = reply_to_id

        logger.debug("[%s] Dispatching message from %s", self.name, user_name)
        await self.handle_message(message_event)

    # ------------------------------------------------------------------
    # Outbound messaging — send responses to Zulip
    # ------------------------------------------------------------------

    async def send(
        self,
        chat_id: str,
        content: str,
        reply_to: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> SendResult:
        """Send a message to a Zulip chat.

        For DMs, chat_id is the user_id. For streams, format is
        "stream_name:topic". Sends a placeholder first if this is the
        first message in a turn, then edits it on subsequent calls.

        Accepts **kwargs for Hermes Gateway compatibility.
        """
        # Determine target type and IDs
        is_dm = not chat_id or ":" not in chat_id
        truncated = _truncate(content)

        if is_dm:
            # DM: chat_id is the user_id
            to = [int(chat_id)]
            payload = {
                "type": "private",
                "to": json.dumps(to),
                "content": truncated,
            }
        else:
            # Stream: chat_id is "stream_name:topic"
            parts = chat_id.split(":", 1)
            stream_name = parts[0]
            topic = parts[1] if len(parts) > 1 else "general"
            payload = {
                "type": "stream",
                "to": stream_name,
                "subject": topic,
                "content": truncated,
            }

        # Check if we should send a placeholder first (first message in turn)
        placeholder_msg_id = self._pending_replies.pop(chat_id, None)

        if placeholder_msg_id:
            # This is the first response — send placeholder and return its ID
            placeholder = self._pick_placeholder()
            placehold_payload = {**payload, "content": placeholder}
            result = await self._send_api_call(placehold_payload)
            if result and result.get("id"):
                msg_id = str(result["id"])
                logger.debug(
                    "[%s] Sent placeholder msg=%s for %s",
                    self.name, msg_id, chat_id,
                )
                self._health_stats["send_count"] += 1
                return SendResult(
                    success=True,
                    message_id=msg_id,
                )

        # Send actual content
        result = await self._send_api_call(payload)
        if result and result.get("id"):
            msg_id = str(result["id"])
            self._health_stats["send_count"] += 1
            return SendResult(
                success=True,
                message_id=msg_id,
            )

        self._health_stats["send_errors"] += 1
        return SendResult(
            success=False,
            error="Failed to send message",
        )

    async def edit_message(
        self,
        chat_id: str,
        message_id: str,
        content: str,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> SendResult:
        """Edit a previously sent Zulip message (for placeholder replacement).

        Returns SendResult for Hermes Gateway streaming interface compatibility.
        Accepts **kwargs (e.g., finalize=True) — extra kwargs are silently ignored.

        Note: Zulip uses POST /api/v1/messages/{message_id} for editing.
        """
        truncated = _truncate(content)
        payload = {
            "content": truncated,
        }
        try:
            # Zulip API: PATCH /api/v1/messages/{message_id} with content in body
            resp, status = await self._api_call(
                "PATCH", f"/api/v1/messages/{int(message_id)}",
                data=payload,
            )
            if resp:
                self._health_stats["send_count"] += 1
                return SendResult(
                    success=True,
                    message_id=str(resp.get("id", message_id)),
                )
            self._health_stats["send_errors"] += 1
            return SendResult(
                success=False,
                error=f"Edit failed (status={status})",
            )
        except Exception as e:
            logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e)
            self._health_stats["send_errors"] += 1
            return SendResult(
                success=False,
                error=str(e),
            )

    async def send_typing(self, chat_id: str, metadata: Optional[Dict] = None) -> None:
        """Send typing indicator to a Zulip chat."""
        if ":" in chat_id:
            return  # No typing for stream messages
        try:
            user_ids = json.dumps([int(chat_id)])
            await self._api_call(
                "POST", "/api/v1/typing",
                data={"to": user_ids, "op": "start"},
            )
            # typing indicator failure is non-critical
        except Exception:
            pass  # Non-critical

    async def stop_typing(self, chat_id: str) -> None:
        """Stop typing indicator."""
        if ":" in chat_id:
            return
        try:
            user_ids = json.dumps([int(chat_id)])
            await self._api_call(
                "POST", "/api/v1/typing",
                data={"to": user_ids, "op": "stop"},
            )
            # typing indicator failure is non-critical
        except Exception:
            pass  # Non-critical

    async def get_chat_info(self, chat_id: str) -> dict:
        """Return basic info about a Zulip chat."""
        if ":" in chat_id:
            parts = chat_id.split(":", 1)
            return {"name": parts[0], "type": "channel", "topic": parts[1]}
        return {"name": chat_id, "type": "dm"}

    async def delete_message(
        self, chat_id: str, message_id: str,
        metadata: Optional[Dict] = None,
        **kwargs: Any,
    ) -> SendResult:
        """Delete a message via Zulip API."""
        # Zulip doesn't support message deletion via API for bots
        # Send an empty edit instead
        return await self.edit_message(chat_id, message_id, "*deleted*", **kwargs)

    # ------------------------------------------------------------------
    # Gen 2: Self-test diagnostics
    # ------------------------------------------------------------------

    async def selftest(self) -> Dict[str, Any]:
        """Run a self-test to verify adapter health.

        Checks:
        1. Connection state (queue registered, HTTP client alive)
        2. Bot identity resolved (bot_user_id is set)
        3. Queue polling active (poll_task is running)
        4. Echo-loop prevention configured (bot_email matches registered email)
        5. Subscription to primary stream (if reachable)
        6. @all-bots ID resolved
        7. Dedup maintenance task running

        Returns a dict with pass/fail for each check and an overall verdict.
        """
        checks = {}

        # 1. Connection state
        checks["connected"] = {
            "status": self._connected,
            "detail": "Queue registered and connected"
            if self._connected
            else "Not connected to Zulip",
        }

        # 2. Queue registered
        checks["queue_registered"] = {
            "status": self._queue_id is not None,
            "detail": f"Queue: {self._queue_id}"
            if self._queue_id
            else "No queue registered",
        }

        # 3. HTTP client alive
        checks["http_client"] = {
            "status": self._http_client is not None and not self._http_client.is_closed,
            "detail": "HTTP client active"
            if self._http_client and not self._http_client.is_closed
            else "HTTP client unavailable",
        }

        # 4. Bot identity
        checks["bot_identity"] = {
            "status": self._bot_user_id is not None,
            "detail": f"Bot user_id: {self._bot_user_id}"
            if self._bot_user_id
            else "Bot user_id not resolved",
        }

        # 5. Poll loop active
        checks["poll_loop"] = {
            "status": self._poll_task is not None and not self._poll_task.done(),
            "detail": "Poll loop running"
            if self._poll_task and not self._poll_task.done()
            else "Poll loop not active",
        }

        # 6. Dedup cleanup active
        checks["dedup_cleanup"] = {
            "status": self._dedup_cleanup_task is not None
            and not self._dedup_cleanup_task.done(),
            "detail": "Dedup maintenance running"
            if self._dedup_cleanup_task and not self._dedup_cleanup_task.done()
            else "Dedup maintenance not active",
        }

        # 7. Echo-loop prevention
        checks["echo_prevention"] = {
            "status": bool(self._bot_email),
            "detail": f"Bot email: {self._bot_email}",
        }

        # 8. @all-bots configured
        checks["all_bots_configured"] = {
            "status": self._all_bots_user_id is not None,
            "detail": f"@all-bots user_id: {self._all_bots_user_id}",
        }

        # Overall verdict
        critical = ["connected", "queue_registered", "http_client", "poll_loop"]
        passed = sum(1 for c in checks.values() if c["status"])
        critical_passed = sum(1 for k in critical if checks.get(k, {}).get("status"))
        failed = [k for k, v in checks.items() if not v["status"]]

        if critical_passed == len(critical) and passed == len(checks):
            verdict = "healthy"
        elif critical_passed < len(critical):
            verdict = "critical_failure"
        else:
            verdict = "degraded"

        return {
            "verdict": verdict,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "checks": checks,
            "summary": {
                "total": len(checks),
                "passed": passed,
                "failed": len(failed),
                "failed_checks": failed,
                "critical_passed": critical_passed,
                "critical_total": len(critical),
            },
        }

    # ------------------------------------------------------------------
    # Gen 3: Health stats callback
    # ------------------------------------------------------------------

    def set_health_callback(
        self,
        callback,
        interval: int = 600,
    ) -> None:
        """Register a callback for periodic health stats reporting.

        The callback is called with health stats dict every `interval`
        seconds. Can be used by the Hermes Gateway to log health to
        RA-H OS knowledge graph or external monitoring.

        Args:
            callback: Async callable receiving health stats dict
            interval: Seconds between calls (default 600 = 10 min)
        """
        self._health_callback = callback
        self._health_callback_interval = interval

    async def _report_health_if_callback(self) -> None:
        """Call the health stats callback if one is registered."""
        if self._health_callback:
            try:
                stats = await self.get_health_stats()
                await self._health_callback(stats)
            except Exception as e:
                logger.debug(
                    "[%s] Health callback error: %s", self.name, e
                )

    # ------------------------------------------------------------------
    # Gen 2: Health stats
    # ------------------------------------------------------------------

    async def get_health_stats(self) -> Dict[str, Any]:
        """Return health and performance statistics.

        Returns a snapshot of:
        - Uptime
        - Poll counts, errors, reconnects
        - Message counts (total, DM, mention)
        - Send counts and errors
        - Dedup map size
        - Timestamps of last activity

        Suitable for periodic logging to RA-H OS knowledge graph.
        """
        now = datetime.now(timezone.utc)

        uptime = None
        if self._health_stats.get("started_at"):
            try:
                started = datetime.fromisoformat(self._health_stats["started_at"])
                uptime_seconds = (now - started).total_seconds()
                uptime = f"{uptime_seconds:.0f}s"
            except (ValueError, TypeError):
                pass

        stats = dict(self._health_stats)
        stats["uptime"] = uptime
        stats["dedup_map_size"] = len(self._seen_message_ids)
        stats["queue_id"] = self._queue_id
        stats["bot_user_id"] = self._bot_user_id
        stats["all_bots_user_id"] = self._all_bots_user_id
        stats["connected"] = self._connected
        stats["checked_at"] = now.isoformat()
        stats["total_polls"] = self._total_poll_count
        stats["consecutive_empty_polls"] = self._consecutive_empty_polls
        stats["client_pool_resets"] = self._client_pool_reset_count
        stats["silence_seconds"] = round(time.time() - self._last_event_received_time)

        # Compute error rate
        total_polls = stats.get("poll_count", 0) or 1
        stats["error_rate"] = round(
            stats.get("poll_errors", 0) / total_polls, 4
        )

        return stats

    # ------------------------------------------------------------------
    # Zulip API helpers
    # ------------------------------------------------------------------

    async def _api_call(
        self,
        method: str,
        path: str,
        data: Optional[Dict] = None,
        params: Optional[Dict] = None,
    ) -> Tuple[Optional[Dict], int]:
        """Make an API call to Zulip.

        Returns (response_json, status_code). status_code is useful for
        callers that need to detect specific HTTP errors like 400
        (BAD_EVENT_QUEUE_ID). Returns (None, 0) on transport errors.
        """
        if not self._http_client:
            return None, 0

        url = f"{self._site}{path}"
        headers = {
            "Authorization": self._auth_header,
        }

        try:
            if method == "GET":
                response = await self._http_client.get(
                    url, params=params, headers=headers,
                )
            elif method == "POST":
                response = await self._http_client.post(
                    url, data=data, headers=headers,
                )
            elif method == "PATCH":
                response = await self._http_client.patch(
                    url, data=data, headers=headers,
                )
            else:
                return None, 0

            if response.status_code >= 400:
                # Gen 4: Truncate error bodies to prevent HTML spam
                body = response.text[:MAX_ERROR_LOG_LEN].replace("\n", " ")
                logger.warning(
                    "[%s] API %s %s: %d %s",
                    self.name, method, path,
                    response.status_code, body,
                )
                return None, response.status_code

            return response.json(), response.status_code

        except httpx.TimeoutException:
            logger.debug("[%s] Timeout on %s %s", self.name, method, path)
            return None, 0
        except Exception as e:
            logger.warning("[%s] API error %s %s: %s", self.name, method, path, e)
            return None, 0

    async def _send_api_call(self, payload: Dict) -> Optional[Dict]:
        """Send a message to Zulip."""
        result, _ = await self._api_call("POST", "/api/v1/messages", data=payload)
        return result

    # ------------------------------------------------------------------
    # Utilities
    # ------------------------------------------------------------------

    def _pick_placeholder(self) -> str:
        """Pick a random placeholder message for streaming UX."""
        idx = hash(str(time.time())) % len(PLACEHOLDERS)
        return PLACEHOLDERS[idx]

    def _is_duplicate(self, msg_id: str) -> bool:
        """Deduplication using message IDs — O(1) lookup, no cleanup.

        Cleanup runs in a separate background task (_dedup_cleanup_forever)
        so this path stays fast.
        """
        if msg_id in self._seen_message_ids:
            return True
        self._seen_message_ids[msg_id] = time.time()
        return False

    def get_all_bots_user_id(self) -> int:
        """Return the current @all-bots user ID.

        This is used by the zulip-mention-reliability contract for
        cross-contract verification.
        """
        return self._all_bots_user_id

    def get_bot_user_id(self) -> Optional[int]:
        """Return the bot's own user ID for identity checks."""
        return self._bot_user_id


# ------------------------------------------------------------------
# Plugin registration helpers
# ------------------------------------------------------------------

def check_requirements() -> bool:
    """Check whether the Zulip adapter is installable."""
    if not HTTPX_AVAILABLE:
        return False
    site = os.getenv("ZULIP_SITE", "").strip() or os.getenv("ZULIP_URL", "").strip()
    email = os.getenv("ZULIP_EMAIL", "").strip()
    api_key = os.getenv("ZULIP_API_KEY", "").strip()
    return bool(site and email and api_key)


def validate_config(config) -> bool:
    """Validate that the configured Zulip platform has credentials."""
    extra = getattr(config, "extra", {}) or {}
    site = extra.get("site") or os.getenv("ZULIP_SITE", "") or os.getenv("ZULIP_URL", "")
    email = extra.get("email") or os.getenv("ZULIP_EMAIL", "")
    api_key = extra.get("api_key") or os.getenv("ZULIP_API_KEY", "")
    return bool(site and email and api_key)


def is_connected(config) -> bool:
    """Check whether Zulip is configured (env or config.yaml)."""
    return validate_config(config)


def _env_enablement() -> Optional[dict]:
    """Seeds PlatformConfig.extra from env vars for env-only setups."""
    site = os.getenv("ZULIP_SITE", "").strip() or os.getenv("ZULIP_URL", "").strip()
    email = os.getenv("ZULIP_EMAIL", "").strip()
    api_key = os.getenv("ZULIP_API_KEY", "").strip()
    if not (site and email and api_key):
        return None
    result: Dict[str, Any] = {
        "site": site,
        "email": email,
        "api_key": api_key,
    }
    stream = os.getenv("ZULIP_STREAM", "").strip()
    if stream:
        result["stream"] = stream
    agent_name = os.getenv("ZULIP_AGENT_NAME", "").strip()
    if agent_name:
        result["agent_name"] = agent_name
    home = os.getenv("ZULIP_HOME_CHANNEL", "").strip()
    if home:
        result["home_channel"] = home
    return result


def register(ctx) -> None:
    """Plugin entry point — called by the Hermes plugin system at startup."""
    ctx.register_platform(
        name="zulip",
        label="Zulip",
        adapter_factory=lambda cfg: ZulipAdapter(cfg),
        check_fn=check_requirements,
        validate_config=validate_config,
        is_connected=is_connected,
        required_env=["ZULIP_SITE", "ZULIP_EMAIL", "ZULIP_API_KEY"],
        install_hint="pip install httpx   # already a Hermes dependency",
        env_enablement_fn=_env_enablement,
        cron_deliver_env_var="ZULIP_HOME_CHANNEL",
        allowed_users_env="ZULIP_ALLOWED_USERS",
        allow_all_env="ZULIP_ALLOW_ALL_USERS",
        max_message_length=MAX_ZULIP_MESSAGE,
        emoji="💬",
        pii_safe=True,
        allow_update_command=True,
        platform_hint=(
            "You are communicating via Zulip. Format responses with "
            "Zulip-compatible Markdown. Keep responses concise and "
            "well-structured. DMs are private conversations; stream "
            "messages are visible to all subscribers."
        ),
    )