diff --git a/hermes_cli/main.py b/hermes_cli/main.py
index c233594fcf8..2dd7e2b603f 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -14002,6 +14002,24 @@ def main():
                 seen_plugin_commands.add(cmd_info["name"])
 
             discover_plugins()
+
+            # Bundled platform plugins register LAZILY (perf, #54448): their
+            # modules — and therefore their register_cli_command() calls —
+            # only run when the platform_registry is asked for them. When the
+            # user invokes `hermes <platform> ...` (e.g. `hermes photon`,
+            # `hermes xchat`), resolve that one deferred loader now so the
+            # plugin's CLI subcommand exists in the parser. Cheap: imports a
+            # single platform module, and only on an unknown-subcommand path.
+            _first_pos = _first_positional_argv()
+            if _first_pos and _first_pos not in seen_plugin_commands:
+                try:
+                    from gateway.platform_registry import platform_registry
+
+                    if platform_registry.is_registered(_first_pos):
+                        platform_registry.get(_first_pos)  # fires the loader
+                except Exception:
+                    pass
+
             for cmd_info in get_plugin_manager()._cli_commands.values():
                 if cmd_info["name"] in seen_plugin_commands:
                     continue
diff --git a/plugins/platforms/xchat/__init__.py b/plugins/platforms/xchat/__init__.py
new file mode 100644
index 00000000000..742d10a6eb3
--- /dev/null
+++ b/plugins/platforms/xchat/__init__.py
@@ -0,0 +1,4 @@
+"""X Chat (end-to-end encrypted X DMs) platform plugin entry point."""
+from .adapter import register
+
+__all__ = ["register"]
diff --git a/plugins/platforms/xchat/adapter.py b/plugins/platforms/xchat/adapter.py
new file mode 100644
index 00000000000..77c42a0b8d7
--- /dev/null
+++ b/plugins/platforms/xchat/adapter.py
@@ -0,0 +1,770 @@
+"""X Chat platform adapter (Hermes plugin).
+
+Connects the Hermes gateway to X's end-to-end encrypted direct messages
+via the official X Chat API. All plaintext stays local: inbound
+``encoded_event`` blobs are decrypted with the Chat XDK (``chatxdk``) and
+outbound replies are encrypted + signed before they ever reach X.
+
+Transport model
+---------------
+Inbound is a polling loop over ``GET /2/chat/conversations/{id}/events``
+(the same shape as X's own bot example). Conversations are auto-discovered
+via ``GET /2/chat/conversations`` (or pinned with
+``XCHAT_CONVERSATION_IDS``); each is polled every ``XCHAT_POLL_INTERVAL``
+seconds with exponential backoff on errors. Outbound goes through
+``POST /2/chat/conversations/{id}/messages``.
+
+Identity / key state (written by ``hermes xchat setup``):
+
+* ``XCHAT_ACCESS_TOKEN``   OAuth2 user token (dm.read, dm.write, users.read, tweet.read)
+* ``XCHAT_USER_ID``        the bot account's numeric user id
+* ``XCHAT_SIGNING_KEY_VERSION``  registered public-key version
+* private-key blob at ``~/.hermes/xchat/private_keys.b64`` (mode 600), or
+  ``XCHAT_PRIVATE_KEYS_B64`` env override
+
+The E2EE session is one ``chat_xdk.Chat`` instance with ``set_identity`` +
+``set_cache_keys(True)``: KeyChange events route through the batch decrypt
+path to feed the verified-key cache, so encrypt calls need no explicit
+conversation key.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    SendResult,
+)
+
+from .api import HTTPX_AVAILABLE, XChatApi, XChatApiError, XChatRateLimited
+from .crypto import XChatCrypto, message_text
+
+logger = logging.getLogger(__name__)
+
+# X DMs cap out around 10k chars; stay under it so chunking kicks in first.
+MAX_MESSAGE_LENGTH = 9500
+
+DEFAULT_POLL_INTERVAL = 10.0
+DISCOVERY_INTERVAL = 300.0  # re-list conversations every 5 minutes
+ERROR_BACKOFF = [5, 15, 30, 60, 120]
+DEDUP_MAX_SIZE = 5000
+
+# Group-chat mention wake words — same defaults as the other Hermes channels
+# so group gating behaves identically everywhere.
+_DEFAULT_MENTION_PATTERNS = [
+    r"(?<![\w@])@?hermes\s+agent\b[,:\-]?",
+    r"(?<![\w@])@?hermes\b[,:\-]?",
+]
+
+
+def _state_dir() -> Path:
+    from hermes_constants import get_hermes_home
+
+    return get_hermes_home() / "xchat"
+
+
+def _read_key_blob() -> str:
+    """Private-key blob: env override first, then the setup-written file."""
+    env_blob = os.getenv("XCHAT_PRIVATE_KEYS_B64", "").strip()
+    if env_blob:
+        return env_blob
+    blob_path = _state_dir() / "private_keys.b64"
+    try:
+        return blob_path.read_text(encoding="utf-8").strip()
+    except OSError:
+        return ""
+
+
+def check_requirements() -> bool:
+    """True when the adapter is minimally configured (token + key material).
+
+    Deliberately does NOT import chatxdk — the native SDK lazy-installs at
+    connect time; a pre-flight check must stay cheap.
+    """
+    if not HTTPX_AVAILABLE:
+        return False
+    if not os.getenv("XCHAT_ACCESS_TOKEN", "").strip():
+        return False
+    return bool(_read_key_blob())
+
+
+def validate_config(config) -> bool:
+    extra = getattr(config, "extra", {}) or {}
+    token = extra.get("access_token") or os.getenv("XCHAT_ACCESS_TOKEN", "")
+    return bool(token)
+
+
+def is_connected(config) -> bool:
+    extra = getattr(config, "extra", {}) or {}
+    token = os.getenv("XCHAT_ACCESS_TOKEN") or extra.get("access_token", "")
+    return bool(token)
+
+
+def _compile_mention_patterns(raw: Any) -> List[re.Pattern]:
+    """Accept list / JSON string / comma- or newline-separated string / None."""
+    patterns: List[str]
+    if raw is None or raw == "":
+        patterns = _DEFAULT_MENTION_PATTERNS
+    elif isinstance(raw, list):
+        patterns = [str(p) for p in raw if str(p).strip()]
+    else:
+        text = str(raw).strip()
+        if text.startswith("["):
+            try:
+                patterns = [str(p) for p in json.loads(text)]
+            except (ValueError, TypeError):
+                patterns = [text]
+        else:
+            parts = re.split(r"[\n,]+", text)
+            patterns = [p.strip() for p in parts if p.strip()]
+        if not patterns:
+            patterns = _DEFAULT_MENTION_PATTERNS
+    compiled = []
+    for p in patterns:
+        try:
+            compiled.append(re.compile(p, re.IGNORECASE))
+        except re.error:
+            logger.warning("[xchat] invalid mention pattern skipped: %r", p)
+    return compiled or [re.compile(p, re.IGNORECASE) for p in _DEFAULT_MENTION_PATTERNS]
+
+
+class XChatAdapter(BasePlatformAdapter):
+    """X Chat (encrypted X DMs) adapter."""
+
+    MAX_MESSAGE_LENGTH = MAX_MESSAGE_LENGTH
+
+    def __init__(self, config: PlatformConfig):
+        platform = Platform("xchat")
+        super().__init__(config=config, platform=platform)
+
+        extra = config.extra or {}
+        self._access_token: str = (
+            extra.get("access_token") or os.getenv("XCHAT_ACCESS_TOKEN", "")
+        ).strip()
+        self._refresh_token: str = (
+            extra.get("refresh_token") or os.getenv("XCHAT_REFRESH_TOKEN", "")
+        ).strip()
+        self._client_id: str = (
+            extra.get("client_id") or os.getenv("XCHAT_CLIENT_ID", "")
+        ).strip()
+        self._client_secret: str = (
+            extra.get("client_secret") or os.getenv("XCHAT_CLIENT_SECRET", "")
+        ).strip()
+        self._bot_user_id: str = str(
+            extra.get("user_id") or os.getenv("XCHAT_USER_ID", "")
+        ).strip()
+        self._signing_key_version: str = str(
+            extra.get("signing_key_version")
+            or os.getenv("XCHAT_SIGNING_KEY_VERSION", "1")
+        ).strip() or "1"
+
+        try:
+            self._poll_interval = float(
+                extra.get("poll_interval") or os.getenv("XCHAT_POLL_INTERVAL", "") or DEFAULT_POLL_INTERVAL
+            )
+        except (TypeError, ValueError):
+            self._poll_interval = DEFAULT_POLL_INTERVAL
+        self._poll_interval = max(2.0, self._poll_interval)
+
+        # Pinned conversations (skip discovery when set).
+        conv_raw = extra.get("conversation_ids") or os.getenv("XCHAT_CONVERSATION_IDS", "")
+        if isinstance(conv_raw, list):
+            self._pinned_conversations = [str(c).strip() for c in conv_raw if str(c).strip()]
+        else:
+            self._pinned_conversations = [
+                c.strip() for c in str(conv_raw).split(",") if c.strip()
+            ]
+
+        # Group mention gating.
+        env_require = os.getenv("XCHAT_REQUIRE_MENTION")
+        if env_require is not None:
+            self.require_mention = env_require.strip().lower() in {"1", "true", "yes"}
+        else:
+            self.require_mention = bool(extra.get("require_mention", False))
+        self._mention_patterns = _compile_mention_patterns(
+            extra.get("mention_patterns") or os.getenv("XCHAT_MENTION_PATTERNS")
+        )
+
+        # Runtime state
+        self._api: Optional[XChatApi] = None
+        self._crypto: Optional[XChatCrypto] = None
+        self._poll_task: Optional[asyncio.Task] = None
+        self._running = False
+        self._lock_acquired = False
+
+        # Per-conversation cursors + dedup
+        self._conversations: Set[str] = set(self._pinned_conversations)
+        self._backlog_loaded: Set[str] = set()
+        self._seen_event_ids: Dict[str, float] = {}
+        self._conversation_keys: Dict[str, Dict[str, bytes]] = {}
+        self._last_event_id: Dict[str, str] = {}
+
+        # Signing-key roster (accumulated; the SDK store is replaced wholesale)
+        self._signing_keys: List[Dict[str, str]] = []
+        self._known_senders: Set[str] = set()
+
+        logger.info(
+            "[xchat] adapter initialized: user_id=%s poll=%.0fs pinned=%d refresh=%s",
+            self._bot_user_id or "?",
+            self._poll_interval,
+            len(self._pinned_conversations),
+            "yes" if (self._refresh_token and self._client_id) else "no",
+        )
+
+    # -- Connection lifecycle -------------------------------------------------
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        if not HTTPX_AVAILABLE:
+            logger.warning("[xchat] httpx not installed")
+            return False
+        if not self._access_token:
+            logger.warning("[xchat] XCHAT_ACCESS_TOKEN not configured — run `hermes xchat setup`")
+            return False
+
+        key_blob = _read_key_blob()
+        if not key_blob:
+            logger.warning(
+                "[xchat] no private-key blob found (~/.hermes/xchat/private_keys.b64 "
+                "or XCHAT_PRIVATE_KEYS_B64) — run `hermes xchat setup`"
+            )
+            return False
+
+        # One credential = one gateway. Prevents two profiles polling (and
+        # double-replying) on the same bot account.
+        try:
+            from gateway.status import acquire_scoped_lock
+
+            ok, holder = acquire_scoped_lock("xchat", self._access_token[:16])
+            if not ok:
+                logger.error("[xchat] credential already in use by another gateway: %s", holder)
+                return False
+            self._lock_acquired = True
+        except Exception:
+            logger.debug("[xchat] scoped lock unavailable; continuing", exc_info=True)
+
+        self._api = XChatApi(
+            self._access_token,
+            refresh_token=self._refresh_token,
+            client_id=self._client_id,
+            client_secret=self._client_secret,
+            on_token_refresh=self._persist_rotated_tokens,
+        )
+
+        # Derive the bot's own user id when not configured.
+        if not self._bot_user_id:
+            try:
+                me = await self._api.get_my_user()
+                self._bot_user_id = str(me.get("id") or "")
+            except XChatApiError as e:
+                logger.error("[xchat] failed to resolve bot user id: %s", e)
+                await self._teardown()
+                return False
+        if not self._bot_user_id:
+            logger.error("[xchat] could not determine bot user id")
+            await self._teardown()
+            return False
+
+        # Unlock the E2EE session. chatxdk lazy-installs here on first use.
+        try:
+            crypto = XChatCrypto()
+            crypto.load_keys(key_blob, self._signing_key_version)
+            crypto.set_identity(self._bot_user_id)
+            crypto.set_cache_keys(True)
+            self._crypto = crypto
+        except Exception as e:
+            logger.error("[xchat] failed to initialize Chat XDK session: %s", e)
+            await self._teardown()
+            return False
+
+        self._running = True
+        self._poll_task = asyncio.create_task(self._run_poll_loop())
+        self._mark_connected()
+        logger.info("[xchat] connected as user %s", self._bot_user_id)
+        return True
+
+    async def disconnect(self) -> None:
+        self._running = False
+        if self._poll_task is not None:
+            self._poll_task.cancel()
+            try:
+                await self._poll_task
+            except (asyncio.CancelledError, Exception):
+                pass
+            self._poll_task = None
+        await self._teardown()
+        logger.info("[xchat] disconnected")
+
+    async def _teardown(self) -> None:
+        if self._api is not None:
+            await self._api.aclose()
+            self._api = None
+        if self._lock_acquired:
+            try:
+                from gateway.status import release_scoped_lock
+
+                release_scoped_lock("xchat", self._access_token[:16])
+            except Exception:
+                pass
+            self._lock_acquired = False
+
+    async def _persist_rotated_tokens(self, access_token: str, refresh_token: str) -> None:
+        """X rotates the refresh token on every renewal — persist both to .env."""
+        self._access_token = access_token
+        self._refresh_token = refresh_token
+        try:
+            from hermes_cli.config import save_env_value
+
+            save_env_value("XCHAT_ACCESS_TOKEN", access_token)
+            if refresh_token:
+                save_env_value("XCHAT_REFRESH_TOKEN", refresh_token)
+        except Exception:
+            logger.warning("[xchat] failed to persist rotated OAuth tokens", exc_info=True)
+
+    # -- Polling loop -----------------------------------------------------------
+
+    async def _run_poll_loop(self) -> None:
+        backoff_idx = 0
+        last_discovery = 0.0
+        while self._running:
+            try:
+                now = time.monotonic()
+                if not self._pinned_conversations and (
+                    now - last_discovery >= DISCOVERY_INTERVAL or not self._conversations
+                ):
+                    await self._discover_conversations()
+                    last_discovery = now
+
+                for conv_id in list(self._conversations):
+                    if not self._running:
+                        return
+                    await self._poll_conversation(conv_id)
+
+                backoff_idx = 0
+                await asyncio.sleep(self._poll_interval)
+            except asyncio.CancelledError:
+                return
+            except XChatRateLimited as e:
+                wait = max(30.0, (e.reset_epoch - time.time()) if e.reset_epoch else 60.0)
+                wait = min(wait, 900.0)
+                logger.warning("[xchat] rate limited — sleeping %.0fs", wait)
+                await asyncio.sleep(wait)
+            except Exception as e:
+                delay = ERROR_BACKOFF[min(backoff_idx, len(ERROR_BACKOFF) - 1)]
+                backoff_idx += 1
+                logger.warning("[xchat] poll error (retry in %ds): %s", delay, e)
+                await asyncio.sleep(delay)
+
+    async def _discover_conversations(self) -> None:
+        assert self._api is not None
+        token: Optional[str] = None
+        found: Set[str] = set()
+        for _ in range(10):  # hard page cap
+            page = await self._api.get_conversations(max_results=100, pagination_token=token)
+            for conv in page.get("data") or []:
+                cid = str(conv.get("conversation_id") or conv.get("id") or "").strip()
+                if cid:
+                    found.add(cid)
+            token = (page.get("meta") or {}).get("next_token")
+            if not token:
+                break
+        new = found - self._conversations
+        if new:
+            logger.info("[xchat] discovered %d new conversation(s)", len(new))
+        self._conversations |= found
+
+    async def _poll_conversation(self, conv_id: str) -> None:
+        assert self._api is not None and self._crypto is not None
+        page = await self._api.get_events(conv_id, max_results=50)
+        raw = page.get("data") or []
+        if not raw:
+            return
+
+        # Events arrive newest-first; process oldest-first.
+        raw = list(reversed(raw))
+        await self._register_signing_keys(raw)
+
+        if conv_id not in self._backlog_loaded:
+            # First sight of this conversation: batch-decrypt to seed the
+            # SDK's verified-key cache, but do NOT reply to the backlog.
+            events_b64 = [e["encoded_event"] for e in raw if e.get("encoded_event")]
+            if events_b64:
+                try:
+                    batch = self._crypto.decrypt_batch(events_b64)
+                    keys = (batch.get("conversation_keys") or {}).get("keys") or {}
+                    self._conversation_keys.setdefault(conv_id, {}).update(keys)
+                except Exception as e:
+                    logger.warning("[xchat] backlog decrypt failed conv=%s: %s", conv_id, e)
+            for item in raw:
+                eid = str(item.get("id") or "")
+                if eid:
+                    self._seen_event_ids[eid] = time.time()
+            self._backlog_loaded.add(conv_id)
+            return
+
+        for item in raw:
+            event_id = str(item.get("id") or "")
+            if not event_id or event_id in self._seen_event_ids:
+                continue
+            self._seen_event_ids[event_id] = time.time()
+            self._prune_dedup()
+
+            event_b64 = item.get("encoded_event")
+            if not event_b64:
+                continue
+            try:
+                event = self._crypto.decrypt_one(
+                    event_b64, self._conversation_keys.get(conv_id) or None
+                )
+            except Exception as e:
+                logger.warning("[xchat] decrypt failed conv=%s event=%s: %s", conv_id, event_id, e)
+                continue
+
+            etype = event.get("type")
+            if etype == "KeyChange":
+                # Key rotation: route through the batch path — it verifies the
+                # change and feeds the SDK's verified-key cache.
+                try:
+                    rotated = self._crypto.decrypt_batch([event_b64])
+                    keys = (rotated.get("conversation_keys") or {}).get("keys") or {}
+                    self._conversation_keys.setdefault(conv_id, {}).update(keys)
+                except Exception as e:
+                    logger.warning("[xchat] key-change processing failed conv=%s: %s", conv_id, e)
+                continue
+            if etype != "Message":
+                continue
+
+            sender_id = str(event.get("sender_id") or item.get("sender_id") or "")
+            if sender_id == self._bot_user_id:
+                continue  # echo of our own reply
+
+            text = message_text(event)
+            if not text:
+                continue
+
+            # The signature covers the canonical conversation id embedded in
+            # the event — prefer it for replies.
+            canonical_conv = str(event.get("conversation_id") or conv_id)
+            await self._dispatch_inbound(
+                conv_id=canonical_conv,
+                sender_id=sender_id,
+                text=text,
+                message_id=event_id,
+                raw=item,
+            )
+
+    def _prune_dedup(self) -> None:
+        if len(self._seen_event_ids) <= DEDUP_MAX_SIZE:
+            return
+        # Drop the oldest half.
+        items = sorted(self._seen_event_ids.items(), key=lambda kv: kv[1])
+        for eid, _ in items[: len(items) // 2]:
+            self._seen_event_ids.pop(eid, None)
+
+    async def _register_signing_keys(self, events: List[Dict[str, Any]]) -> None:
+        """Fetch new senders' public keys into the SDK's signing-key store."""
+        assert self._api is not None and self._crypto is not None
+        senders = {
+            str(e.get("sender_id"))
+            for e in events
+            if e.get("sender_id") and str(e.get("sender_id")) != self._bot_user_id
+        } - self._known_senders
+        for sender_id in senders:
+            try:
+                for pk in await self._api.get_public_keys(sender_id):
+                    self._signing_keys.append(
+                        {
+                            "user_id": sender_id,
+                            "public_key_version": str(pk.get("public_key_version") or ""),
+                            "public_key": pk.get("signing_public_key") or "",
+                            "identity_public_key": pk.get("public_key") or "",
+                            "identity_public_key_signature": pk.get("identity_public_key_signature") or "",
+                        }
+                    )
+                self._known_senders.add(sender_id)
+            except Exception:
+                logger.warning("[xchat] public-key fetch failed sender=%s", sender_id)
+        if senders and self._signing_keys:
+            # The SDK store is replaced wholesale — push the full roster.
+            self._crypto.set_signing_keys(self._signing_keys)
+
+    # -- Inbound dispatch --------------------------------------------------------
+
+    def _message_matches_mention_patterns(self, text: str) -> bool:
+        return any(p.search(text) for p in self._mention_patterns)
+
+    def _clean_mention_text(self, text: str) -> str:
+        """Strip ONLY a leading wake-word match — never mid-prompt words."""
+        stripped = text.lstrip()
+        for p in self._mention_patterns:
+            m = p.match(stripped)
+            if m:
+                return stripped[m.end():].lstrip()
+        return text
+
+    async def _dispatch_inbound(
+        self,
+        *,
+        conv_id: str,
+        sender_id: str,
+        text: str,
+        message_id: str,
+        raw: Dict[str, Any],
+    ) -> None:
+        is_group = conv_id.startswith("g")
+        chat_type = "group" if is_group else "dm"
+
+        if is_group and self.require_mention:
+            if not self._message_matches_mention_patterns(text):
+                return
+            text = self._clean_mention_text(text)
+            if not text:
+                return
+
+        source = self.build_source(
+            chat_id=conv_id,
+            chat_name=None,
+            chat_type=chat_type,
+            user_id=sender_id,
+            user_name=None,
+            message_id=message_id,
+        )
+        event = MessageEvent(
+            text=text,
+            message_type=MessageType.TEXT,
+            source=source,
+            raw_message=raw,
+            message_id=message_id,
+        )
+        await self.handle_message(event)
+
+    # -- Outbound ------------------------------------------------------------------
+
+    async def send(
+        self,
+        chat_id: str,
+        content: str,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        if self._api is None or self._crypto is None:
+            return SendResult(success=False, error="xchat adapter not connected")
+        if len(content) > MAX_MESSAGE_LENGTH:
+            content = content[:MAX_MESSAGE_LENGTH]
+        try:
+            body = self._crypto.encrypt_text(chat_id, content)
+        except ValueError:
+            # No verified conversation key cached yet. For a 1:1, the key
+            # cache seeds from the conversation backlog; a brand-new
+            # conversation the bot initiates needs a key-change first —
+            # out of scope for reply flows (the poll loop always seeds
+            # keys before we ever reply).
+            return SendResult(
+                success=False,
+                error=(
+                    "No verified conversation key for this conversation yet. "
+                    "The key cache seeds from inbound events — reply flows "
+                    "always have it; initiating brand-new conversations is "
+                    "not supported yet."
+                ),
+            )
+        except Exception as e:
+            return SendResult(success=False, error=f"encrypt failed: {e}")
+        try:
+            out = await self._api.send_message(chat_id, body)
+        except XChatApiError as e:
+            logger.warning("[xchat] send failed conv=%s: %s", chat_id, e)
+            return SendResult(success=False, error=str(e))
+        data = out.get("data") or {}
+        msg_id = str(data.get("message_id") or body.get("message_id") or "")
+        # Suppress the echo when it comes back around the poll loop.
+        for eid_key in ("event_id", "id"):
+            eid = data.get(eid_key)
+            if eid:
+                self._seen_event_ids[str(eid)] = time.time()
+        return SendResult(success=True, message_id=msg_id)
+
+    async def send_typing(self, chat_id: str, metadata=None) -> None:
+        if self._api is None:
+            return
+        try:
+            await self._api.send_typing(chat_id)
+        except Exception:
+            pass  # best-effort
+
+    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+        chat_type = "group" if str(chat_id).startswith("g") else "dm"
+        return {"name": str(chat_id), "type": chat_type, "chat_id": str(chat_id)}
+
+
+# ---------------------------------------------------------------------------
+# Plugin registration
+
+
+def _env_enablement() -> Optional[dict]:
+    """Seed ``PlatformConfig.extra`` from env vars during gateway config load."""
+    token = os.getenv("XCHAT_ACCESS_TOKEN", "").strip()
+    if not token:
+        return None
+    seed: dict = {"access_token": token}
+    for env, key in (
+        ("XCHAT_REFRESH_TOKEN", "refresh_token"),
+        ("XCHAT_CLIENT_ID", "client_id"),
+        ("XCHAT_CLIENT_SECRET", "client_secret"),
+        ("XCHAT_USER_ID", "user_id"),
+        ("XCHAT_SIGNING_KEY_VERSION", "signing_key_version"),
+        ("XCHAT_CONVERSATION_IDS", "conversation_ids"),
+        ("XCHAT_POLL_INTERVAL", "poll_interval"),
+    ):
+        val = os.getenv(env, "").strip()
+        if val:
+            seed[key] = val
+    home = os.getenv("XCHAT_HOME_CHANNEL", "").strip()
+    if home:
+        seed["home_channel"] = {
+            "chat_id": home,
+            "name": os.getenv("XCHAT_HOME_CHANNEL_NAME", home),
+        }
+    return seed
+
+
+async def _standalone_send(
+    pconfig,
+    chat_id: str,
+    message: str,
+    *,
+    thread_id: Optional[str] = None,
+    media_files: Optional[List[Any]] = None,
+    force_document: bool = False,
+) -> Dict[str, Any]:
+    """Out-of-process encrypted send for cron / send_message_tool.
+
+    Opens an ephemeral API client + Chat XDK session, seeds the
+    conversation key from the conversation's event backlog, encrypts,
+    sends, and closes. ``thread_id`` / ``media_files`` are accepted for
+    signature parity — X Chat has no thread primitive and media requires
+    the full streaming-encrypt flow (not wired yet).
+    """
+    if not HTTPX_AVAILABLE:
+        return {"error": "xchat standalone send: httpx not installed"}
+
+    extra = getattr(pconfig, "extra", {}) or {}
+    access_token = (extra.get("access_token") or os.getenv("XCHAT_ACCESS_TOKEN", "")).strip()
+    if not access_token:
+        return {"error": "xchat standalone send: XCHAT_ACCESS_TOKEN not configured"}
+    key_blob = _read_key_blob()
+    if not key_blob:
+        return {"error": "xchat standalone send: private-key blob missing — run `hermes xchat setup`"}
+    user_id = str(extra.get("user_id") or os.getenv("XCHAT_USER_ID", "")).strip()
+    key_version = str(
+        extra.get("signing_key_version") or os.getenv("XCHAT_SIGNING_KEY_VERSION", "1")
+    ).strip() or "1"
+
+    api = XChatApi(
+        access_token,
+        refresh_token=(extra.get("refresh_token") or os.getenv("XCHAT_REFRESH_TOKEN", "")).strip(),
+        client_id=(extra.get("client_id") or os.getenv("XCHAT_CLIENT_ID", "")).strip(),
+        client_secret=(extra.get("client_secret") or os.getenv("XCHAT_CLIENT_SECRET", "")).strip(),
+    )
+    try:
+        if not user_id:
+            me = await api.get_my_user()
+            user_id = str(me.get("id") or "")
+        if not user_id:
+            return {"error": "xchat standalone send: could not resolve bot user id"}
+
+        crypto = XChatCrypto()
+        crypto.load_keys(key_blob, key_version)
+        crypto.set_identity(user_id)
+        crypto.set_cache_keys(True)
+
+        # Seed the conversation key from the backlog (KeyChange events).
+        page = await api.get_events(chat_id, max_results=50)
+        events_b64 = [e["encoded_event"] for e in (page.get("data") or []) if e.get("encoded_event")]
+        canonical = chat_id
+        if events_b64:
+            try:
+                batch = crypto.decrypt_batch(events_b64)
+                for m in batch.get("messages") or []:
+                    conv = (m.get("event") or {}).get("conversation_id")
+                    if conv:
+                        canonical = str(conv)
+                        break
+            except Exception as e:
+                logger.debug("[xchat] standalone backlog decrypt: %s", e)
+
+        try:
+            body = crypto.encrypt_text(canonical, message)
+        except ValueError:
+            return {
+                "error": (
+                    "xchat: no verified conversation key — the target must have "
+                    "an existing conversation with the bot"
+                )
+            }
+        out = await api.send_message(canonical, body)
+        data = out.get("data") or {}
+        return {
+            "success": True,
+            "platform": "xchat",
+            "chat_id": canonical,
+            "message_id": str(data.get("message_id") or body.get("message_id") or ""),
+        }
+    except XChatApiError as e:
+        return {"error": f"xchat standalone send failed: {e}"}
+    except Exception as e:
+        return {"error": f"xchat standalone send failed: {e}"}
+    finally:
+        await api.aclose()
+
+
+def register(ctx) -> None:
+    """Plugin entry point — called by the Hermes plugin loader at startup."""
+    from . import cli as _cli
+
+    ctx.register_platform(
+        name="xchat",
+        label="X Chat (encrypted DMs)",
+        adapter_factory=lambda cfg: XChatAdapter(cfg),
+        check_fn=check_requirements,
+        validate_config=validate_config,
+        is_connected=is_connected,
+        required_env=["XCHAT_ACCESS_TOKEN"],
+        install_hint=(
+            "Run: hermes xchat setup  (stores the OAuth2 user token, registers "
+            "the bot's E2EE keys, saves the private-key blob)."
+        ),
+        setup_fn=_cli.gateway_setup,
+        env_enablement_fn=_env_enablement,
+        cron_deliver_env_var="XCHAT_HOME_CHANNEL",
+        standalone_sender_fn=_standalone_send,
+        allowed_users_env="XCHAT_ALLOWED_USERS",
+        allow_all_env="XCHAT_ALLOW_ALL_USERS",
+        max_message_length=MAX_MESSAGE_LENGTH,
+        emoji="𝕏",
+        pii_safe=False,
+        allow_update_command=True,
+        platform_hint=(
+            "You are communicating via X Chat — X's end-to-end encrypted "
+            "direct messages. Treat replies like regular chat messages: "
+            "short and conversational. Markdown is NOT rendered — use plain "
+            "text. User identifiers are numeric X user ids; conversation ids "
+            "starting with 'g' are group chats."
+        ),
+    )
+
+    ctx.register_cli_command(
+        name="xchat",
+        help="Set up and manage the X Chat (encrypted X DMs) integration",
+        setup_fn=_cli.register_cli,
+        handler_fn=_cli.dispatch,
+    )
diff --git a/plugins/platforms/xchat/api.py b/plugins/platforms/xchat/api.py
new file mode 100644
index 00000000000..4e374fd54da
--- /dev/null
+++ b/plugins/platforms/xchat/api.py
@@ -0,0 +1,278 @@
+"""Async X API client for the X Chat platform adapter.
+
+A thin httpx-based client for the handful of X API v2 endpoints the adapter
+needs. The official ``xdk`` Python client is synchronous (requests-based),
+which doesn't fit the async gateway — the Chat endpoints are plain
+OAuth2-bearer REST, so direct calls are wire-identical. Only the E2EE layer
+needs a real SDK (``chatxdk``, see ``crypto.py``).
+
+Also owns OAuth2 token refresh: X user access tokens expire (~2h). When a
+refresh token + client id are configured, :meth:`XChatApi.ensure_token`
+renews the access token through ``POST /2/oauth2/token`` and persists the
+rotated pair via a caller-supplied callback (X rotates refresh tokens on
+every use).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any, Awaitable, Callable, Optional
+
+try:
+    import httpx
+    HTTPX_AVAILABLE = True
+except ImportError:  # pragma: no cover - httpx is a core Hermes dependency
+    HTTPX_AVAILABLE = False
+    httpx = None  # type: ignore[assignment]
+
+logger = logging.getLogger(__name__)
+
+BASE_URL = "https://api.x.com"
+
+# Fields we always request on the events endpoint — the decrypt path needs
+# encoded_event; sender_id/conversation_id drive session routing.
+_EVENT_FIELDS = (
+    "conversation_id,created_at_msec,encoded_event,id,sender_id"
+)
+
+# Refresh the access token this many seconds before its reported expiry.
+_TOKEN_REFRESH_SLACK = 300
+
+
+class XChatApiError(Exception):
+    """Raised for non-2xx responses from the X API."""
+
+    def __init__(self, status: int, detail: str) -> None:
+        self.status = status
+        self.detail = detail
+        super().__init__(f"X API HTTP {status}: {detail}")
+
+
+class XChatRateLimited(XChatApiError):
+    """HTTP 429 — carries the reset epoch when the API reports one."""
+
+    def __init__(self, detail: str, reset_epoch: Optional[int]) -> None:
+        super().__init__(429, detail)
+        self.reset_epoch = reset_epoch
+
+
+class XChatApi:
+    """Async client bound to one bot account's OAuth2 user token."""
+
+    def __init__(
+        self,
+        access_token: str,
+        *,
+        refresh_token: str = "",
+        client_id: str = "",
+        client_secret: str = "",
+        token_expires_at: float = 0.0,
+        on_token_refresh: Optional[Callable[[str, str], Awaitable[None]]] = None,
+        base_url: str = BASE_URL,
+        client: Optional["httpx.AsyncClient"] = None,
+    ) -> None:
+        self._access_token = access_token
+        self._refresh_token = refresh_token
+        self._client_id = client_id
+        self._client_secret = client_secret
+        # 0 = unknown expiry; refresh only reactively on 401.
+        self._token_expires_at = token_expires_at
+        self._on_token_refresh = on_token_refresh
+        self._base_url = base_url.rstrip("/")
+        self._client = client
+        self._refresh_lock = asyncio.Lock()
+
+    # -- lifecycle -----------------------------------------------------------
+
+    def _http(self) -> "httpx.AsyncClient":
+        if self._client is None:
+            self._client = httpx.AsyncClient(timeout=30.0)
+        return self._client
+
+    async def aclose(self) -> None:
+        if self._client is not None:
+            try:
+                await self._client.aclose()
+            except Exception:
+                pass
+            self._client = None
+
+    # -- auth ----------------------------------------------------------------
+
+    @property
+    def can_refresh(self) -> bool:
+        return bool(self._refresh_token and self._client_id)
+
+    async def ensure_token(self) -> None:
+        """Proactively refresh the access token when close to expiry."""
+        if not self.can_refresh or not self._token_expires_at:
+            return
+        if time.time() < self._token_expires_at - _TOKEN_REFRESH_SLACK:
+            return
+        await self._refresh_access_token()
+
+    async def _refresh_access_token(self) -> None:
+        """POST /2/oauth2/token (refresh_token grant). Rotates both tokens."""
+        async with self._refresh_lock:
+            # Another task may have refreshed while we waited on the lock.
+            if self._token_expires_at and time.time() < self._token_expires_at - _TOKEN_REFRESH_SLACK:
+                return
+            data = {
+                "grant_type": "refresh_token",
+                "refresh_token": self._refresh_token,
+                "client_id": self._client_id,
+            }
+            auth = None
+            if self._client_secret:
+                auth = (self._client_id, self._client_secret)
+            resp = await self._http().post(
+                f"{self._base_url}/2/oauth2/token", data=data, auth=auth
+            )
+            if resp.status_code >= 300:
+                raise XChatApiError(resp.status_code, resp.text[:300])
+            tok = resp.json()
+            self._access_token = tok.get("access_token") or self._access_token
+            # X rotates refresh tokens on every use — always adopt the new one.
+            new_refresh = tok.get("refresh_token")
+            if new_refresh:
+                self._refresh_token = new_refresh
+            expires_in = tok.get("expires_in")
+            if expires_in:
+                self._token_expires_at = time.time() + float(expires_in)
+            logger.info("[xchat] OAuth2 access token refreshed")
+            if self._on_token_refresh is not None:
+                try:
+                    await self._on_token_refresh(self._access_token, self._refresh_token)
+                except Exception:
+                    logger.warning("[xchat] token persist callback failed", exc_info=True)
+
+    # -- request core ----------------------------------------------------------
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Optional[dict[str, Any]] = None,
+        json_body: Optional[dict[str, Any]] = None,
+        _retried_auth: bool = False,
+    ) -> dict[str, Any]:
+        await self.ensure_token()
+        headers = {"Authorization": f"Bearer {self._access_token}"}
+        resp = await self._http().request(
+            method,
+            f"{self._base_url}{path}",
+            params=params,
+            json=json_body,
+            headers=headers,
+        )
+        if resp.status_code == 401 and self.can_refresh and not _retried_auth:
+            # Reactive refresh — covers the no-known-expiry case.
+            await self._refresh_access_token()
+            return await self._request(
+                method, path, params=params, json_body=json_body, _retried_auth=True
+            )
+        if resp.status_code == 429:
+            reset = resp.headers.get("x-user-limit-24hour-reset") or resp.headers.get(
+                "x-rate-limit-reset"
+            )
+            raise XChatRateLimited(
+                resp.text[:300], int(reset) if reset and reset.isdigit() else None
+            )
+        if resp.status_code >= 300:
+            raise XChatApiError(resp.status_code, resp.text[:300])
+        if not resp.content:
+            return {}
+        try:
+            return resp.json()
+        except ValueError:
+            return {}
+
+    # -- identity ------------------------------------------------------------
+
+    async def get_my_user(self) -> dict[str, Any]:
+        """GET /2/users/me — the bot account's own id/username."""
+        out = await self._request("GET", "/2/users/me")
+        return out.get("data") or {}
+
+    async def get_public_keys(self, user_id: str) -> list[dict[str, Any]]:
+        """GET /2/users/{id}/public_keys — a user's registered E2EE keys."""
+        out = await self._request("GET", f"/2/users/{user_id}/public_keys")
+        data = out.get("data") or []
+        return data if isinstance(data, list) else [data]
+
+    async def add_public_key(self, user_id: str, body: dict[str, Any]) -> dict[str, Any]:
+        """POST /2/users/{id}/public_keys — register the bot's public keys.
+
+        Rate limited to a handful of writes per 24h; raises
+        :class:`XChatRateLimited` on 429 so callers stop instead of burning
+        the daily budget.
+        """
+        return await self._request("POST", f"/2/users/{user_id}/public_keys", json_body=body)
+
+    # -- conversations ---------------------------------------------------------
+
+    @staticmethod
+    def _conv_path_id(conversation_id: str) -> str:
+        # Events embed the colon form; URL paths take the hyphen form.
+        return str(conversation_id).replace(":", "-")
+
+    async def get_conversations(
+        self, *, max_results: int = 100, pagination_token: Optional[str] = None
+    ) -> dict[str, Any]:
+        """GET /2/chat/conversations — list the bot's conversations."""
+        params: dict[str, Any] = {"max_results": max_results}
+        if pagination_token:
+            params["pagination_token"] = pagination_token
+        return await self._request("GET", "/2/chat/conversations", params=params)
+
+    async def get_events(
+        self,
+        conversation_id: str,
+        *,
+        max_results: int = 50,
+        pagination_token: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """GET /2/chat/conversations/{id}/events — raw (encrypted) events."""
+        params: dict[str, Any] = {
+            "max_results": max_results,
+            "chat_message_event.fields": _EVENT_FIELDS,
+        }
+        if pagination_token:
+            params["pagination_token"] = pagination_token
+        return await self._request(
+            "GET",
+            f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/events",
+            params=params,
+        )
+
+    async def send_message(self, conversation_id: str, body: dict[str, str]) -> dict[str, Any]:
+        """POST /2/chat/conversations/{id}/messages — send encrypted ciphertext.
+
+        ``body`` is the dict produced by ``XChatCrypto.encrypt_text``. For a
+        1:1 conversation ``conversation_id`` may be the recipient's bare user
+        id; the server derives the canonical conversation id.
+        """
+        return await self._request(
+            "POST",
+            f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/messages",
+            json_body=body,
+        )
+
+    async def send_typing(self, conversation_id: str) -> None:
+        """POST /2/chat/conversations/{id}/typing — best-effort typing indicator."""
+        await self._request(
+            "POST",
+            f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/typing",
+        )
+
+    async def add_conversation_keys(
+        self, conversation_id: str, body: dict[str, Any]
+    ) -> dict[str, Any]:
+        """POST /2/chat/conversations/{id}/keys — initialize/rotate a conversation key."""
+        return await self._request(
+            "POST",
+            f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/keys",
+            json_body=body,
+        )
diff --git a/plugins/platforms/xchat/cli.py b/plugins/platforms/xchat/cli.py
new file mode 100644
index 00000000000..563c9bd81fb
--- /dev/null
+++ b/plugins/platforms/xchat/cli.py
@@ -0,0 +1,335 @@
+"""``hermes xchat ...`` CLI subcommands — registered by the plugin via
+``ctx.register_cli_command()``.
+
+Subcommands:
+
+    setup       full first-time setup (token + user id + key generation/registration)
+    register    (re)register the E2EE public keys only
+    status      show token / key / registration state
+
+Key registration is a rare, rate-limited write (only a few per 24h per
+account). ``setup`` is safe to re-run: the private-key blob and the
+registration payload are persisted to ``~/.hermes/xchat/`` BEFORE any
+network call, so an interrupted run resumes the same identity instead of
+minting a new one and burning the daily budget.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+from hermes_cli.colors import Colors, color
+
+
+def _state_dir() -> Path:
+    from hermes_constants import get_hermes_home
+
+    d = get_hermes_home() / "xchat"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _blob_path() -> Path:
+    return _state_dir() / "private_keys.b64"
+
+
+def _marker_path() -> Path:
+    return _state_dir() / "registration.json"
+
+
+def _read_marker() -> dict:
+    try:
+        return json.loads(_marker_path().read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+
+
+def _write_marker(marker: dict) -> None:
+    _marker_path().write_text(json.dumps(marker, indent=2) + "\n", encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# argparse wiring
+
+
+def register_cli(parser: argparse.ArgumentParser) -> None:
+    """Wire up `hermes xchat ...` subcommands."""
+    subs = parser.add_subparsers(dest="xchat_command", required=False)
+
+    p_setup = subs.add_parser(
+        "setup",
+        help="First-time setup (OAuth token + E2EE key generation/registration)",
+    )
+    p_setup.add_argument(
+        "--force",
+        action="store_true",
+        help="Generate a NEW identity even if one is already registered (dangerous)",
+    )
+
+    subs.add_parser("register", help="(Re)register the E2EE public keys with the X API")
+    subs.add_parser("status", help="Show token / key / registration state")
+
+    parser.set_defaults(func=dispatch)
+
+
+def dispatch(args: argparse.Namespace) -> int:
+    sub = getattr(args, "xchat_command", None)
+    if sub in (None, "status"):
+        return cmd_status()
+    if sub == "setup":
+        return cmd_setup(force=getattr(args, "force", False))
+    if sub == "register":
+        return cmd_register(force=False)
+    print(color(f"Unknown xchat subcommand: {sub}", Colors.RED))
+    return 1
+
+
+def gateway_setup() -> None:
+    """Zero-arg hook for the unified `hermes gateway setup` wizard."""
+    cmd_setup(force=False)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+
+
+def _get_env(key: str) -> str:
+    val = os.getenv(key, "").strip()
+    if val:
+        return val
+    # Fall back to the persisted .env (the CLI may run before env load).
+    try:
+        from hermes_cli.config import get_env_value
+
+        return (get_env_value(key) or "").strip()
+    except Exception:
+        return ""
+
+
+def _save_env(key: str, value: str) -> None:
+    from hermes_cli.config import save_env_value
+
+    save_env_value(key, value)
+    os.environ[key] = value
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+async def _fetch_user_id(api) -> str:
+    me = await api.get_my_user()
+    return str(me.get("id") or "")
+
+
+# ---------------------------------------------------------------------------
+# Commands
+
+
+def cmd_status() -> int:
+    token = _get_env("XCHAT_ACCESS_TOKEN")
+    user_id = _get_env("XCHAT_USER_ID")
+    marker = _read_marker()
+    blob = _blob_path()
+
+    print(color("X Chat integration status", Colors.BOLD))
+    print(f"  access token:  {'✓ set' if token else '✗ missing'}")
+    print(f"  refresh token: {'✓ set' if _get_env('XCHAT_REFRESH_TOKEN') else '– not set (no auto-renew)'}")
+    print(f"  bot user id:   {user_id or '– not set'}")
+    print(f"  key blob:      {'✓ ' + str(blob) if blob.exists() else '✗ missing'}")
+    if marker.get("registered"):
+        print(f"  registration:  ✓ version {marker.get('version')} ({marker.get('registered_at', '?')})")
+    else:
+        print("  registration:  ✗ not registered — run `hermes xchat setup`")
+    return 0
+
+
+def cmd_setup(*, force: bool) -> int:
+    print(color("X Chat setup — end-to-end encrypted X DMs", Colors.BOLD))
+    print(
+        "You need an X developer app with OAuth 2.0 user-context enabled and a\n"
+        "user access token carrying: dm.read dm.write users.read tweet.read\n"
+        "(offline.access too if you want refresh tokens).\n"
+        "Docs: https://docs.x.com/xchat/getting-started\n"
+    )
+
+    # 1. Token
+    token = _get_env("XCHAT_ACCESS_TOKEN")
+    if token:
+        print("Found existing XCHAT_ACCESS_TOKEN.")
+    else:
+        import getpass
+
+        token = getpass.getpass("Paste the OAuth2 user access token: ").strip()
+        if not token:
+            print(color("No token provided — aborting.", Colors.RED))
+            return 1
+        _save_env("XCHAT_ACCESS_TOKEN", token)
+    refresh = _get_env("XCHAT_REFRESH_TOKEN")
+    if not refresh:
+        import getpass
+
+        refresh = getpass.getpass(
+            "Paste the OAuth2 refresh token (optional, Enter to skip): "
+        ).strip()
+        if refresh:
+            _save_env("XCHAT_REFRESH_TOKEN", refresh)
+            client_id = input("X app OAuth2 client id (needed for refresh): ").strip()
+            if client_id:
+                _save_env("XCHAT_CLIENT_ID", client_id)
+
+    # 2. Bot user id
+    from .api import XChatApi, XChatApiError
+
+    api = XChatApi(token)
+    user_id = _get_env("XCHAT_USER_ID")
+    if not user_id:
+        try:
+            user_id = _run(_fetch_user_id(api))
+        except XChatApiError as e:
+            print(color(f"Could not resolve the bot's user id: {e}", Colors.RED))
+            print("Check the token's scopes (users.read) and validity.")
+            return 1
+        finally:
+            _run(api.aclose())
+            api = None
+        if not user_id:
+            print(color("Could not resolve the bot's user id.", Colors.RED))
+            return 1
+        _save_env("XCHAT_USER_ID", user_id)
+        print(f"Bot user id: {user_id}")
+    else:
+        _run(api.aclose())
+        api = None
+
+    # 3. Keys + registration
+    rc = cmd_register(force=force)
+    if rc != 0:
+        return rc
+
+    print()
+    print(color("Setup complete.", Colors.GREEN))
+    print("Enable the platform and start the gateway:")
+    print("  hermes gateway start")
+    return 0
+
+
+def cmd_register(*, force: bool) -> int:
+    """Generate + register the E2EE keys. Re-runnable / resume-safe."""
+    token = _get_env("XCHAT_ACCESS_TOKEN")
+    if not token:
+        print(color("XCHAT_ACCESS_TOKEN not set — run `hermes xchat setup` first.", Colors.RED))
+        return 1
+    user_id = _get_env("XCHAT_USER_ID")
+
+    from .api import XChatApi, XChatApiError, XChatRateLimited
+    from .crypto import XChatCrypto
+
+    marker = _read_marker()
+    if marker.get("registered") and not force:
+        print(
+            f"Already registered (key version {marker.get('version')}). "
+            "Use `hermes xchat setup --force` to mint a NEW identity."
+        )
+        return 0
+
+    async def _register() -> int:
+        nonlocal user_id
+        api = XChatApi(
+            token,
+            refresh_token=_get_env("XCHAT_REFRESH_TOKEN"),
+            client_id=_get_env("XCHAT_CLIENT_ID"),
+            client_secret=_get_env("XCHAT_CLIENT_SECRET"),
+        )
+        try:
+            if not user_id:
+                user_id = await _fetch_user_id(api)
+                if not user_id:
+                    print(color("Could not resolve the bot's user id.", Colors.RED))
+                    return 1
+                _save_env("XCHAT_USER_ID", user_id)
+
+            crypto = XChatCrypto()
+            blob_path = _blob_path()
+            resuming = blob_path.exists() and marker.get("body") and not force
+            if resuming:
+                crypto.load_keys(blob_path.read_text(encoding="utf-8").strip())
+                body = marker["body"]
+                version = str(marker.get("version") or "1")
+                print(f"Resuming the saved identity ({blob_path}).")
+            else:
+                payload = crypto.generate_and_register_payload()
+                body = payload["registration"]
+                version = payload["version"]
+                blob_path.write_text(payload["private_keys_b64"] + "\n", encoding="utf-8")
+                try:
+                    blob_path.chmod(0o600)
+                except OSError:
+                    pass
+                _write_marker(
+                    {"registered": False, "user_id": user_id, "version": version, "body": body}
+                )
+                print(f"Generated a new identity; private keys saved to {blob_path} (mode 600).")
+
+            our_public_key = body["public_key"]["public_key"]
+
+            # Reconcile: adopt an already-registered key instead of re-POSTing
+            # (a prior POST may have applied server-side after erroring).
+            try:
+                existing = await api.get_public_keys(user_id)
+            except XChatApiError:
+                existing = []
+            already = next(
+                (k for k in existing if k.get("public_key") == our_public_key), None
+            )
+            if already:
+                version = str(already.get("public_key_version") or version)
+                print(f"Public key already registered (version {version}); skipping POST.")
+            else:
+                print(f"Registering public key version {version} …")
+                try:
+                    resp = await api.add_public_key(user_id, body)
+                except XChatRateLimited as limited:
+                    when = (
+                        datetime.fromtimestamp(limited.reset_epoch, tz=timezone.utc).isoformat()
+                        if limited.reset_epoch
+                        else "the next window"
+                    )
+                    print(
+                        color(
+                            "Registration is rate limited (429). The daily budget is "
+                            f"exhausted; wait until {when} and re-run — the saved "
+                            "identity resumes, so no budget is wasted.",
+                            Colors.RED,
+                        )
+                    )
+                    return 1
+                data = resp.get("data") or {}
+                if isinstance(data, list):
+                    data = data[0] if data else {}
+                version = str(data.get("public_key_version") or version)
+
+            _save_env("XCHAT_SIGNING_KEY_VERSION", version)
+            _write_marker(
+                {
+                    "registered": True,
+                    "user_id": user_id,
+                    "version": version,
+                    "body": body,
+                    "registered_at": datetime.now(timezone.utc).isoformat(),
+                }
+            )
+            print(color(f"Key registration complete (version {version}).", Colors.GREEN))
+            return 0
+        finally:
+            await api.aclose()
+
+    try:
+        return _run(_register())
+    except XChatApiError as e:
+        print(color(f"Registration failed: {e}", Colors.RED))
+        return 1
diff --git a/plugins/platforms/xchat/crypto.py b/plugins/platforms/xchat/crypto.py
new file mode 100644
index 00000000000..640322ccbe4
--- /dev/null
+++ b/plugins/platforms/xchat/crypto.py
@@ -0,0 +1,169 @@
+"""Crypto core for the X Chat platform adapter.
+
+A thin, network-free wrapper around the ``chat_xdk`` binding. Everything
+that touches the Chat XDK lives here so it can be unit-tested with a fake
+``Chat`` object and so the adapter/API layers stay import-light. The SDK is
+lazy-installed at first use via ``tools.lazy_deps`` (feature key
+``platform.xchat``).
+
+Responsibilities:
+
+* key management        -> :meth:`XChatCrypto.load_keys` /
+                           :meth:`XChatCrypto.generate_and_register_payload`
+* session identity      -> :meth:`XChatCrypto.set_identity`
+* signing-key roster    -> :meth:`XChatCrypto.set_signing_keys`
+* message encryption    -> :meth:`XChatCrypto.encrypt_text`
+* event decryption      -> :meth:`XChatCrypto.decrypt_batch` (decrypt_events)
+                           and :meth:`XChatCrypto.decrypt_one` (decrypt_event)
+
+The decrypted-event dict shape follows the Chat XDK: ``{"type": "Message",
+"id": ..., "sender_id": ..., "conversation_id": ..., "content": {"text":
+...}}`` for messages, ``{"type": "KeyChange", ...}`` for conversation-key
+rotations.
+"""
+from __future__ import annotations
+
+import base64
+import logging
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def _as_dict(obj: Any) -> dict[str, Any]:
+    """Decrypted events come back as native objects; normalise to a dict."""
+    if isinstance(obj, dict):
+        return obj
+    if hasattr(obj, "model_dump"):
+        return obj.model_dump()
+    try:
+        return dict(obj)
+    except Exception:
+        return {}
+
+
+def _load_chat_class():
+    """Import (lazy-installing if needed) and return ``chat_xdk.Chat``."""
+    try:
+        from chat_xdk import Chat  # type: ignore[import-not-found]
+        return Chat
+    except ImportError:
+        pass
+    # Lazy-install path — same pattern as the telegram/matrix platform plugins.
+    from tools.lazy_deps import ensure as _lazy_ensure
+
+    _lazy_ensure("platform.xchat", prompt=False)
+    from chat_xdk import Chat  # type: ignore[import-not-found]
+    return Chat
+
+
+class XChatCrypto:
+    """Wraps a single unlocked ``chat_xdk.Chat`` instance for one bot identity."""
+
+    def __init__(self, chat: Any = None) -> None:
+        # ``chat`` injection keeps unit tests free of the native SDK.
+        self.chat = chat if chat is not None else _load_chat_class()()
+        self.signing_key_version: str = "1"
+        self._identity_set = False
+
+    # -- Key management -----------------------------------------------------
+
+    def load_keys(self, private_keys_b64: str, signing_key_version: str = "1") -> None:
+        """Import an existing private-key blob (from ``export_keys``) and adopt it.
+
+        ``private_keys_b64`` is the base64 blob produced during registration
+        (``hermes xchat setup``). Raises on a malformed blob.
+        """
+        blob = base64.b64decode(private_keys_b64.strip())
+        self.chat.import_keys(blob, version=signing_key_version)
+        self.signing_key_version = str(signing_key_version)
+
+    def set_identity(self, user_id: str) -> None:
+        """Set the session identity — every later encrypt call signs as this user."""
+        self.chat.set_identity(str(user_id), self.signing_key_version)
+        self._identity_set = True
+
+    def set_cache_keys(self, enabled: bool = True) -> None:
+        """Opt in to the SDK's verified conversation-key cache."""
+        self.chat.set_cache_keys(enabled)
+
+    def set_signing_keys(self, signing_keys: list[dict[str, str]]) -> None:
+        """Replace the SDK's participant signing-key store (full roster each call)."""
+        self.chat.set_signing_keys(signing_keys)
+
+    def generate_and_register_payload(self) -> dict[str, Any]:
+        """Generate fresh keypairs for a brand-new bot identity.
+
+        Returns the registration body for ``POST /2/users/{id}/public_keys``
+        plus the exported private-key blob (base64) to persist locally.
+        Used by ``hermes xchat setup`` only — the adapter never generates keys.
+        """
+        reg = self.chat.generate_keypairs()
+        version = str(reg.version) if getattr(reg, "version", None) is not None else "1"
+        body = {
+            "public_key": {
+                "public_key": reg.public_key.public_key,
+                "signing_public_key": reg.public_key.signing_public_key,
+                "identity_public_key_signature": reg.public_key.identity_public_key_signature,
+                "signing_public_key_signature": reg.public_key.signing_public_key_signature,
+                "registration_method": reg.public_key.registration_method,
+            },
+            "version": version,
+            "generate_version": bool(getattr(reg, "generate_version", False)),
+        }
+        exported = self.chat.export_keys()
+        blob_b64 = base64.b64encode(bytes(exported)).decode("ascii") if exported else ""
+        return {"registration": body, "version": version, "private_keys_b64": blob_b64}
+
+    # -- Decryption ----------------------------------------------------------
+
+    def decrypt_batch(self, events_b64: list[str]) -> dict[str, Any]:
+        """Batch path — initial backlog load and KeyChange processing.
+
+        ``decrypt_events`` extracts conversation keys from any KeyChange
+        events in the batch (feeding the SDK's key cache when enabled), then
+        decrypts every message. Signing keys come from the
+        ``set_signing_keys`` store.
+        """
+        result = self.chat.decrypt_events(events_b64, None)
+        messages = [
+            {"event": _as_dict(m.get("event") if isinstance(m, dict) else m)}
+            for m in (result.get("messages") or [])
+        ]
+        return {
+            "messages": messages,
+            "conversation_keys": result.get("conversation_keys") or {},
+            "errors": result.get("errors") or {},
+        }
+
+    def decrypt_one(
+        self, event_b64: str, conversation_keys: Optional[dict[str, bytes]] = None
+    ) -> dict[str, Any]:
+        """Single-event path — per-poll decryption with cached conversation keys."""
+        return _as_dict(self.chat.decrypt_event(event_b64, conversation_keys, None))
+
+    # -- Encryption ----------------------------------------------------------
+
+    def encrypt_text(self, conversation_id: str, text: str) -> dict[str, str]:
+        """Encrypt + sign ``text``, returning the X API send-message body.
+
+        The conversation key is resolved from the SDK's verified-key cache
+        (``set_cache_keys``); the sender comes from ``set_identity``. Raises
+        ``ValueError`` when no verified key is cached for the conversation.
+        """
+        payload = self.chat.encrypt_message(str(conversation_id), text)
+        return {
+            "message_id": payload.message_id,
+            "encoded_message_create_event": payload.encrypted_content,
+            "encoded_message_event_signature": payload.encoded_event_signature,
+        }
+
+
+def message_text(event: dict[str, Any]) -> Optional[str]:
+    """Pull the plain text out of a decrypted Message event, or None."""
+    if event.get("type") != "Message":
+        return None
+    content = event.get("content") or {}
+    if isinstance(content, dict):
+        return content.get("text")
+    return None
diff --git a/plugins/platforms/xchat/plugin.yaml b/plugins/platforms/xchat/plugin.yaml
new file mode 100644
index 00000000000..f471858c906
--- /dev/null
+++ b/plugins/platforms/xchat/plugin.yaml
@@ -0,0 +1,80 @@
+name: xchat-platform
+label: X Chat (encrypted DMs)
+kind: platform
+version: 0.1.0
+description: >
+  X Chat gateway adapter for Hermes Agent.
+  Connects the agent to X's end-to-end encrypted direct messages via the
+  official X Chat API. Message bodies are encrypted/decrypted locally with
+  the Chat XDK (chatxdk) — X only ever routes ciphertext. Inbound messages
+  arrive through a polling loop over the conversation-events endpoint;
+  outbound replies are encrypted, signed, and POSTed back.
+
+  One-time onboarding runs through `hermes xchat setup`: it stores the
+  OAuth2 user token, derives the bot's user id, generates + registers the
+  E2EE keypairs, and persists the private-key blob under
+  ``~/.hermes/xchat/`` (mode 600).
+author: NousResearch
+requires_env:
+  - name: XCHAT_ACCESS_TOKEN
+    description: "OAuth2 user access token with dm.read, dm.write, users.read, tweet.read scopes (set by `hermes xchat setup`)"
+    prompt: "X OAuth2 user access token"
+    url: "https://developer.x.com/en/portal/dashboard"
+    password: true
+optional_env:
+  - name: XCHAT_REFRESH_TOKEN
+    description: "OAuth2 refresh token — lets the adapter renew the ~2h access token automatically (requires XCHAT_CLIENT_ID)"
+    prompt: "X OAuth2 refresh token (or empty)"
+    password: true
+  - name: XCHAT_CLIENT_ID
+    description: "OAuth2 app client id — required for automatic token refresh"
+    prompt: "X app OAuth2 client id (or empty)"
+    password: false
+  - name: XCHAT_CLIENT_SECRET
+    description: "OAuth2 app client secret — only for confidential clients"
+    prompt: "X app OAuth2 client secret (or empty)"
+    password: true
+  - name: XCHAT_USER_ID
+    description: "The bot account's numeric X user id (derived automatically by `hermes xchat setup`)"
+    prompt: "Bot X user id (or empty to derive)"
+    password: false
+  - name: XCHAT_SIGNING_KEY_VERSION
+    description: "Registered public-key version for signing (written by `hermes xchat setup`)"
+    prompt: "Signing key version (default 1)"
+    password: false
+  - name: XCHAT_PRIVATE_KEYS_B64
+    description: "Base64 private-key blob from the Chat XDK export — overrides the blob file under ~/.hermes/xchat/"
+    prompt: "Private-key blob (or empty to use the blob file)"
+    password: true
+  - name: XCHAT_ALLOWED_USERS
+    description: "Comma-separated numeric X user ids allowed to talk to the bot"
+    prompt: "Allowed user ids (comma-separated)"
+    password: false
+  - name: XCHAT_ALLOW_ALL_USERS
+    description: "Allow any sender to trigger the bot (dev only — disables allowlist)"
+    prompt: "Allow all users? (true/false)"
+    password: false
+  - name: XCHAT_CONVERSATION_IDS
+    description: "Comma-separated conversation ids to poll (omit to auto-discover all conversations)"
+    prompt: "Pinned conversation ids (or empty)"
+    password: false
+  - name: XCHAT_POLL_INTERVAL
+    description: "Seconds between event polls per cycle (default 10)"
+    prompt: "Poll interval seconds (default 10)"
+    password: false
+  - name: XCHAT_REQUIRE_MENTION
+    description: "Ignore group-chat messages unless they match a mention wake word (true/false, default false)"
+    prompt: "Require a mention in group chats?"
+    password: false
+  - name: XCHAT_MENTION_PATTERNS
+    description: "Mention wake-word regexes for group chats (JSON list or comma/newline-separated; defaults to Hermes wake words)"
+    prompt: "Group mention patterns"
+    password: false
+  - name: XCHAT_HOME_CHANNEL
+    description: "Default X Chat target for cron / notification delivery: conversation id or bare numeric user id"
+    prompt: "Home conversation/user id (or empty)"
+    password: false
+  - name: XCHAT_HOME_CHANNEL_NAME
+    description: "Human label for the home channel (defaults to the id)"
+    prompt: "Home channel display name (or empty)"
+    password: false
diff --git a/pyproject.toml b/pyproject.toml
index ca364827957..777c825404a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -166,6 +166,10 @@ messaging = ["python-telegram-bot[webhooks]==22.6", "discord.py[voice]==2.7.1",
 cron = []  # croniter is now a core dependency; this extra kept for back-compat
 slack = ["slack-bolt==1.29.0", "slack-sdk==3.43.0", "aiohttp==3.14.1"]
 matrix = ["mautrix[encryption]==0.21.0", "aiosqlite==0.22.1", "asyncpg==0.31.0", "aiohttp-socks==0.11.0", "aiohttp==3.14.1"]  # aiohttp 3.14.1: CVE-2026-34993(RCE)/47265 + 34513/34518/34519/34520/34525 (mautrix/aiohttp-socks only cap aiohttp<4 / >=3.10, so pin the patched floor directly)
+# X Chat (encrypted X DMs) adapter — the official Chat XDK E2EE binding.
+# Lazy-installed at first use via tools/lazy_deps.py (platform.xchat);
+# declared here so packagers (Nix, Homebrew) can ship it explicitly.
+xchat = ["chatxdk==0.4.1"]
 # WeCom callback-mode adapter — parses untrusted XML POST bodies from
 # WeCom-controlled callback endpoints, so we use defusedxml (drop-in
 # replacement for stdlib xml.etree.ElementTree) to block billion-laughs
diff --git a/tests/plugins/platforms/xchat/test_xchat_adapter.py b/tests/plugins/platforms/xchat/test_xchat_adapter.py
new file mode 100644
index 00000000000..2833956e51a
--- /dev/null
+++ b/tests/plugins/platforms/xchat/test_xchat_adapter.py
@@ -0,0 +1,528 @@
+"""Unit tests for the X Chat platform plugin.
+
+All tests run offline: the X API layer is replaced with fakes and the Chat
+XDK crypto core is replaced with a stub — no chatxdk native module, no
+network, no gateway process.
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Any, Dict, List, Optional
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent, MessageType
+from plugins.platforms.xchat import adapter as xchat_adapter
+from plugins.platforms.xchat.adapter import (
+    XChatAdapter,
+    _compile_mention_patterns,
+    _env_enablement,
+    check_requirements,
+)
+from plugins.platforms.xchat.crypto import XChatCrypto, message_text
+
+
+# ---------------------------------------------------------------------------
+# Helpers / fakes
+
+
+class FakeCrypto:
+    """Stands in for XChatCrypto — records calls, no native SDK."""
+
+    def __init__(self) -> None:
+        self.encrypted: List[tuple] = []
+        self.batch_calls: List[List[str]] = []
+        self.signing_keys: List[Dict[str, str]] = []
+        self.decrypt_map: Dict[str, Dict[str, Any]] = {}
+        self.fail_encrypt: Optional[Exception] = None
+
+    def decrypt_one(self, event_b64, conversation_keys=None):
+        return self.decrypt_map[event_b64]
+
+    def decrypt_batch(self, events_b64):
+        self.batch_calls.append(list(events_b64))
+        return {"messages": [], "conversation_keys": {"keys": {"1": b"k"}}, "errors": {}}
+
+    def encrypt_text(self, conversation_id, text):
+        if self.fail_encrypt is not None:
+            raise self.fail_encrypt
+        self.encrypted.append((conversation_id, text))
+        return {
+            "message_id": "mid-1",
+            "encoded_message_create_event": "ZW5j",
+            "encoded_message_event_signature": "c2ln",
+        }
+
+    def set_signing_keys(self, keys):
+        self.signing_keys = list(keys)
+
+
+class FakeApi:
+    """Stands in for XChatApi — canned responses, records sends."""
+
+    def __init__(self) -> None:
+        self.sent: List[tuple] = []
+        self.typing: List[str] = []
+        self.public_keys: Dict[str, List[Dict[str, Any]]] = {}
+        self.events_pages: Dict[str, Dict[str, Any]] = {}
+        self.conversations: List[str] = []
+
+    async def get_my_user(self):
+        return {"id": "999"}
+
+    async def get_public_keys(self, user_id):
+        return self.public_keys.get(user_id, [])
+
+    async def get_conversations(self, *, max_results=100, pagination_token=None):
+        return {
+            "data": [{"conversation_id": c} for c in self.conversations],
+            "meta": {},
+        }
+
+    async def get_events(self, conversation_id, *, max_results=50, pagination_token=None):
+        return self.events_pages.get(conversation_id, {"data": []})
+
+    async def send_message(self, conversation_id, body):
+        self.sent.append((conversation_id, body))
+        return {"data": {"message_id": body.get("message_id", ""), "event_id": "evt-echo"}}
+
+    async def send_typing(self, conversation_id):
+        self.typing.append(conversation_id)
+
+    async def aclose(self):
+        pass
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch, **extra) -> XChatAdapter:
+    monkeypatch.setenv("XCHAT_ACCESS_TOKEN", "test-token")
+    monkeypatch.setenv("XCHAT_USER_ID", "999")
+    cfg = PlatformConfig(enabled=True, token="", extra=dict(extra))
+    return XChatAdapter(cfg)
+
+
+def _wire(adapter: XChatAdapter) -> tuple[FakeApi, FakeCrypto]:
+    api, crypto = FakeApi(), FakeCrypto()
+    adapter._api = api
+    adapter._crypto = crypto
+    adapter._bot_user_id = "999"
+    return api, crypto
+
+
+def _capture(adapter: XChatAdapter, monkeypatch: pytest.MonkeyPatch) -> List[MessageEvent]:
+    captured: List[MessageEvent] = []
+
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    return captured
+
+
+# ---------------------------------------------------------------------------
+# check_fn / config
+
+
+def test_check_requirements_needs_token_and_blob(monkeypatch, tmp_path):
+    monkeypatch.delenv("XCHAT_ACCESS_TOKEN", raising=False)
+    monkeypatch.delenv("XCHAT_PRIVATE_KEYS_B64", raising=False)
+    assert check_requirements() is False
+
+    monkeypatch.setenv("XCHAT_ACCESS_TOKEN", "tok")
+    assert check_requirements() is False  # no key blob
+
+    monkeypatch.setenv("XCHAT_PRIVATE_KEYS_B64", "YmxvYg==")
+    assert check_requirements() is True
+
+
+def test_env_enablement_seeds_extra(monkeypatch):
+    monkeypatch.delenv("XCHAT_ACCESS_TOKEN", raising=False)
+    assert _env_enablement() is None
+
+    monkeypatch.setenv("XCHAT_ACCESS_TOKEN", "tok")
+    monkeypatch.setenv("XCHAT_USER_ID", "42")
+    monkeypatch.setenv("XCHAT_HOME_CHANNEL", "123-456")
+    seed = _env_enablement()
+    assert seed is not None
+    assert seed["access_token"] == "tok"
+    assert seed["user_id"] == "42"
+    assert seed["home_channel"]["chat_id"] == "123-456"
+
+
+def test_adapter_reads_config_extra_over_defaults(monkeypatch):
+    adapter = _make_adapter(
+        monkeypatch,
+        poll_interval="30",
+        conversation_ids="111-222, g333",
+    )
+    assert adapter._poll_interval == 30.0
+    assert adapter._pinned_conversations == ["111-222", "g333"]
+    assert adapter.platform == Platform("xchat")
+
+
+def test_poll_interval_floor(monkeypatch):
+    adapter = _make_adapter(monkeypatch, poll_interval="0.1")
+    assert adapter._poll_interval == 2.0
+
+
+# ---------------------------------------------------------------------------
+# Mention gating
+
+
+def test_mention_patterns_json_and_csv():
+    pats = _compile_mention_patterns('["^bot\\\\b"]')
+    assert pats[0].pattern == "^bot\\b"
+    pats = _compile_mention_patterns("alpha, beta")
+    assert len(pats) == 2
+    # None → defaults
+    assert _compile_mention_patterns(None)
+
+
+@pytest.mark.asyncio
+async def test_group_mention_gate(monkeypatch):
+    monkeypatch.setenv("XCHAT_REQUIRE_MENTION", "true")
+    adapter = _make_adapter(monkeypatch)
+    _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+
+    # Group message without wake word → dropped
+    await adapter._dispatch_inbound(
+        conv_id="g123", sender_id="5", text="just chatting", message_id="e1", raw={}
+    )
+    assert captured == []
+
+    # Group message with wake word → dispatched, wake word stripped
+    await adapter._dispatch_inbound(
+        conv_id="g123", sender_id="5", text="hermes what time is it", message_id="e2", raw={}
+    )
+    assert len(captured) == 1
+    assert captured[0].text == "what time is it"
+
+    # DMs are never gated
+    await adapter._dispatch_inbound(
+        conv_id="111-222", sender_id="5", text="no wake word", message_id="e3", raw={}
+    )
+    assert len(captured) == 2
+
+
+@pytest.mark.asyncio
+async def test_dispatch_sets_chat_type(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(
+        conv_id="g42", sender_id="7", text="hi", message_id="e1", raw={}
+    )
+    await adapter._dispatch_inbound(
+        conv_id="111-999", sender_id="7", text="hi", message_id="e2", raw={}
+    )
+    assert captured[0].source.chat_type == "group"
+    assert captured[1].source.chat_type == "dm"
+    assert captured[0].message_type == MessageType.TEXT
+    assert captured[1].source.user_id == "7"
+
+
+# ---------------------------------------------------------------------------
+# Poll loop mechanics
+
+
+@pytest.mark.asyncio
+async def test_backlog_seeds_keys_without_replying(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+
+    api.events_pages["111-999"] = {
+        "data": [
+            {"id": "e1", "encoded_event": "AAA", "sender_id": "111"},
+            {"id": "e2", "encoded_event": "BBB", "sender_id": "111"},
+        ]
+    }
+    await adapter._poll_conversation("111-999")
+
+    # Backlog: batch-decrypted for keys, nothing dispatched, ids marked seen.
+    assert crypto.batch_calls == [["BBB", "AAA"]]  # newest-first reversed
+    assert captured == []
+    assert "e1" in adapter._seen_event_ids and "e2" in adapter._seen_event_ids
+    assert "111-999" in adapter._backlog_loaded
+
+
+@pytest.mark.asyncio
+async def test_new_message_dispatched_after_backlog(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._backlog_loaded.add("111-999")
+
+    crypto.decrypt_map["CCC"] = {
+        "type": "Message",
+        "id": "e3",
+        "sender_id": "111",
+        "conversation_id": "111:999",
+        "content": {"text": "hello agent"},
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e3", "encoded_event": "CCC", "sender_id": "111"}]
+    }
+    await adapter._poll_conversation("111-999")
+
+    assert len(captured) == 1
+    ev = captured[0]
+    assert ev.text == "hello agent"
+    # Reply target uses the canonical id embedded in the signed event.
+    assert ev.source.chat_id == "111:999"
+
+    # Second poll with the same event id → dedup, no double dispatch.
+    await adapter._poll_conversation("111-999")
+    assert len(captured) == 1
+
+
+@pytest.mark.asyncio
+async def test_own_messages_filtered(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._backlog_loaded.add("111-999")
+
+    crypto.decrypt_map["DDD"] = {
+        "type": "Message",
+        "id": "e4",
+        "sender_id": "999",  # the bot itself
+        "content": {"text": "echo of our own reply"},
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e4", "encoded_event": "DDD", "sender_id": "999"}]
+    }
+    await adapter._poll_conversation("111-999")
+    assert captured == []
+
+
+@pytest.mark.asyncio
+async def test_keychange_routes_through_batch(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._backlog_loaded.add("111-999")
+
+    crypto.decrypt_map["KEY"] = {"type": "KeyChange", "id": "e5"}
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e5", "encoded_event": "KEY", "sender_id": "111"}]
+    }
+    await adapter._poll_conversation("111-999")
+
+    assert crypto.batch_calls == [["KEY"]]
+    assert captured == []
+    assert adapter._conversation_keys["111-999"] == {"1": b"k"}
+
+
+@pytest.mark.asyncio
+async def test_signing_keys_fetched_once_per_sender(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    api.public_keys["111"] = [
+        {
+            "public_key_version": "3",
+            "signing_public_key": "SPK",
+            "public_key": "IPK",
+            "identity_public_key_signature": "SIG",
+        }
+    ]
+    events = [{"id": "e1", "sender_id": "111"}]
+    await adapter._register_signing_keys(events)
+    await adapter._register_signing_keys(events)  # second call — cached
+
+    assert len(crypto.signing_keys) == 1
+    entry = crypto.signing_keys[0]
+    assert entry["user_id"] == "111"
+    assert entry["public_key"] == "SPK"
+    assert entry["identity_public_key"] == "IPK"
+
+
+@pytest.mark.asyncio
+async def test_discovery_adds_conversations(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, _ = _wire(adapter)
+    api.conversations = ["111-222", "g333"]
+    await adapter._discover_conversations()
+    assert adapter._conversations == {"111-222", "g333"}
+
+
+def test_dedup_prune_bounds_memory(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    now = time.time()
+    for i in range(xchat_adapter.DEDUP_MAX_SIZE + 100):
+        adapter._seen_event_ids[f"e{i}"] = now + i
+    adapter._prune_dedup()
+    assert len(adapter._seen_event_ids) <= xchat_adapter.DEDUP_MAX_SIZE
+    # Newest entries survive the prune.
+    assert f"e{xchat_adapter.DEDUP_MAX_SIZE + 99}" in adapter._seen_event_ids
+
+
+# ---------------------------------------------------------------------------
+# Outbound
+
+
+@pytest.mark.asyncio
+async def test_send_encrypts_and_posts(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+
+    result = await adapter.send("111:999", "hi there")
+    assert result.success
+    assert result.message_id == "mid-1"
+    assert crypto.encrypted == [("111:999", "hi there")]
+    conv, body = api.sent[0]
+    assert conv == "111:999"
+    assert body["encoded_message_create_event"] == "ZW5j"
+    # Echo suppression: returned event id marked as seen.
+    assert "evt-echo" in adapter._seen_event_ids
+
+
+@pytest.mark.asyncio
+async def test_send_without_conversation_key(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    _, crypto = _wire(adapter)
+    crypto.fail_encrypt = ValueError("no key")
+
+    result = await adapter.send("111:999", "hi")
+    assert not result.success
+    assert "conversation key" in (result.error or "")
+
+
+@pytest.mark.asyncio
+async def test_send_disconnected(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    result = await adapter.send("111:999", "hi")
+    assert not result.success
+
+
+@pytest.mark.asyncio
+async def test_get_chat_info_types(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    assert (await adapter.get_chat_info("g123"))["type"] == "group"
+    assert (await adapter.get_chat_info("111-222"))["type"] == "dm"
+
+
+# ---------------------------------------------------------------------------
+# Crypto wrapper (fake Chat object — no native SDK)
+
+
+class _FakePayload:
+    message_id = "m1"
+    encrypted_content = "ENC"
+    encoded_event_signature = "SIG"
+
+
+class _FakeChat:
+    def __init__(self) -> None:
+        self.identity = None
+        self.cache = None
+        self.imported = None
+
+    def import_keys(self, blob, version=None):
+        self.imported = (blob, version)
+
+    def set_identity(self, user_id, version):
+        self.identity = (user_id, version)
+
+    def set_cache_keys(self, enabled):
+        self.cache = enabled
+
+    def set_signing_keys(self, keys):
+        self.signing = keys
+
+    def encrypt_message(self, conversation_id, text):
+        return _FakePayload()
+
+    def decrypt_event(self, event_b64, conversation_keys, signing_keys):
+        return {"type": "Message", "content": {"text": "plain"}}
+
+    def decrypt_events(self, events, signing_keys):
+        return {"messages": [{"event": {"type": "Message"}}], "conversation_keys": {}, "errors": {}}
+
+
+def test_crypto_load_keys_and_identity():
+    crypto = XChatCrypto(chat=_FakeChat())
+    crypto.load_keys("YmxvYg==", "7")  # b64("blob")
+    assert crypto.chat.imported == (b"blob", "7")
+    assert crypto.signing_key_version == "7"
+    crypto.set_identity("42")
+    assert crypto.chat.identity == ("42", "7")
+
+
+def test_crypto_encrypt_shapes_send_body():
+    crypto = XChatCrypto(chat=_FakeChat())
+    body = crypto.encrypt_text("1:2", "hello")
+    assert body == {
+        "message_id": "m1",
+        "encoded_message_create_event": "ENC",
+        "encoded_message_event_signature": "SIG",
+    }
+
+
+def test_message_text_extraction():
+    assert message_text({"type": "Message", "content": {"text": "x"}}) == "x"
+    assert message_text({"type": "KeyChange"}) is None
+    assert message_text({"type": "Message", "content": {}}) is None
+
+
+# ---------------------------------------------------------------------------
+# Registry integration
+
+
+def test_platform_registry_entry_parity():
+    """Every parity knob must be populated on the registered entry."""
+    from gateway.platform_registry import PlatformEntry
+
+    captured: Dict[str, Any] = {}
+
+    class Ctx:
+        class manifest:
+            name = "xchat-platform"
+
+        def register_platform(self, **kwargs):
+            captured.update(kwargs)
+
+        def register_cli_command(self, **kwargs):
+            captured["cli"] = kwargs
+
+    xchat_adapter.register(Ctx())
+
+    assert captured["name"] == "xchat"
+    assert captured["allowed_users_env"] == "XCHAT_ALLOWED_USERS"
+    assert captured["allow_all_env"] == "XCHAT_ALLOW_ALL_USERS"
+    assert captured["cron_deliver_env_var"] == "XCHAT_HOME_CHANNEL"
+    assert callable(captured["standalone_sender_fn"])
+    assert callable(captured["setup_fn"])
+    assert callable(captured["env_enablement_fn"])
+    assert captured["platform_hint"]
+    assert captured["max_message_length"] > 0
+    assert captured["cli"]["name"] == "xchat"
+    # The kwargs must construct a valid PlatformEntry.
+    entry_kwargs = {k: v for k, v in captured.items() if k != "cli"}
+    entry = PlatformEntry(**entry_kwargs)
+    assert entry.name == "xchat"
+
+
+# ---------------------------------------------------------------------------
+# Standalone send (config-error paths — no network)
+
+
+@pytest.mark.asyncio
+async def test_standalone_send_requires_token(monkeypatch):
+    monkeypatch.delenv("XCHAT_ACCESS_TOKEN", raising=False)
+    monkeypatch.delenv("XCHAT_PRIVATE_KEYS_B64", raising=False)
+    cfg = PlatformConfig(enabled=True, extra={})
+    out = await xchat_adapter._standalone_send(cfg, "111", "msg")
+    assert "XCHAT_ACCESS_TOKEN" in out["error"]
+
+
+@pytest.mark.asyncio
+async def test_standalone_send_requires_key_blob(monkeypatch):
+    monkeypatch.setenv("XCHAT_ACCESS_TOKEN", "tok")
+    monkeypatch.delenv("XCHAT_PRIVATE_KEYS_B64", raising=False)
+    cfg = PlatformConfig(enabled=True, extra={})
+    out = await xchat_adapter._standalone_send(cfg, "111", "msg")
+    assert "private-key blob" in out["error"]
diff --git a/tools/lazy_deps.py b/tools/lazy_deps.py
index ec5692ecd55..a8da267b6fe 100644
--- a/tools/lazy_deps.py
+++ b/tools/lazy_deps.py
@@ -200,6 +200,10 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
     # installed on demand like every other messaging platform; also exposed
     # as the `teams` extra in pyproject for packagers / explicit installs.
     "platform.teams": ("microsoft-teams-apps==2.0.13.4", "aiohttp==3.14.1"),  # aiohttp 3.14.1: CVE-2026-34993(RCE)/47265 + 34513/34518/34519/34520/34525
+    # X Chat (encrypted X DMs) adapter — chatxdk is the official Chat XDK
+    # (native E2EE binding: keygen, encrypt/decrypt, sign/verify). The REST
+    # layer uses core httpx directly, so no xdk client dependency.
+    "platform.xchat": ("chatxdk==0.4.1",),
 
     # ─── Terminal backends ─────────────────────────────────────────────────
     "terminal.modal": ("modal==1.3.4",),
diff --git a/uv.lock b/uv.lock
index 13a9912aa78..7bb54dc3fb0 100644
--- a/uv.lock
+++ b/uv.lock
@@ -673,6 +673,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/0a/4c/925909008ed5a988ccbb72dcc897407e5d6d3bd72410d69e051fc0c14647/charset_normalizer-3.4.4-py3-none-any.whl", hash = "sha256:7a32c560861a02ff789ad905a2fe94e3f840803362c84fecf1851cb4cf3dc37f", size = 53402, upload-time = "2025-10-14T04:42:31.76Z" },
 ]
 
+[[package]]
+name = "chatxdk"
+version = "0.4.1"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/29/2d/38e13cfdf9d0dcee7e13f41015aa4b2648a0f42e0142be12790aa48bdf3e/chatxdk-0.4.1-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:a89d6735417d4c44fbf4d7d7799e58b4cde658536e71eaf37371c9648ef10792", size = 3288306, upload-time = "2026-07-20T23:22:27.254Z" },
+    { url = "https://files.pythonhosted.org/packages/bf/80/81b3bad83a4c986f322d6d591993658c4734a5986453e3186ba85280cf21/chatxdk-0.4.1-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:704c753d474a7c3d3b4dffae0d1a4be3ba78ca911a0b0cc65e85afb3dade37b0", size = 3123290, upload-time = "2026-07-20T23:22:29.043Z" },
+    { url = "https://files.pythonhosted.org/packages/87/68/c0fa19bd67593feb4ffbd965a461bebf685381eeab0ab18551499e740280/chatxdk-0.4.1-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0ed165dffe8c6336d1bbbf6423034e3303c6645c9be72154afae624fb3148e03", size = 3440648, upload-time = "2026-07-20T23:22:30.87Z" },
+    { url = "https://files.pythonhosted.org/packages/9c/3a/975e25b900aa5c1a5d27387f72ac43a5930bc6bbea96f01efc343dcd31f2/chatxdk-0.4.1-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:dba2c5248adaaeafada52980ec739f8068488564a1861d1ca2be0d7a96006a1c", size = 3476274, upload-time = "2026-07-20T23:22:32.728Z" },
+    { url = "https://files.pythonhosted.org/packages/1e/bb/41a3bf9ddd8875a0d3f7445c90324a6039d65bfdd8ddb6de5c29b2139a87/chatxdk-0.4.1-cp310-abi3-win_amd64.whl", hash = "sha256:2b351012cba965e1c2d253381a3ad0e7a6b4858404fc30e811fae70d78b18af3", size = 2959643, upload-time = "2026-07-20T23:22:34.473Z" },
+]
+
 [[package]]
 name = "click"
 version = "8.3.1"
@@ -1727,6 +1739,9 @@ web = [
 wecom = [
     { name = "defusedxml" },
 ]
+xchat = [
+    { name = "chatxdk" },
+]
 youtube = [
     { name = "youtube-transcript-api" },
 ]
@@ -1749,6 +1764,7 @@ requires-dist = [
     { name = "boto3", marker = "extra == 'bedrock'", specifier = "==1.42.89" },
     { name = "brotlicffi", marker = "extra == 'messaging'", specifier = "==1.2.0.1" },
     { name = "certifi", specifier = "==2026.5.20" },
+    { name = "chatxdk", marker = "extra == 'xchat'", specifier = "==0.4.1" },
     { name = "concurrent-log-handler", marker = "sys_platform == 'win32'", specifier = "==0.9.29" },
     { name = "croniter", specifier = "==6.0.0" },
     { name = "cryptography", specifier = "==46.0.7" },
@@ -1855,7 +1871,7 @@ requires-dist = [
     { name = "websockets", specifier = "==15.0.1" },
     { name = "youtube-transcript-api", marker = "extra == 'youtube'", specifier = "==1.2.4" },
 ]
-provides-extras = ["anthropic", "exa", "firecrawl", "parallel-web", "fal", "edge-tts", "modal", "daytona", "hindsight", "dev", "messaging", "cron", "slack", "matrix", "wecom", "cli", "tts-premium", "voice", "pty", "honcho", "supermemory", "mem0", "vision", "mcp", "nemo-relay", "homeassistant", "sms", "teams", "computer-use", "acp", "mistral", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"]
+provides-extras = ["anthropic", "exa", "firecrawl", "parallel-web", "fal", "edge-tts", "modal", "daytona", "hindsight", "dev", "messaging", "cron", "slack", "matrix", "xchat", "wecom", "cli", "tts-premium", "voice", "pty", "honcho", "supermemory", "mem0", "vision", "mcp", "nemo-relay", "homeassistant", "sms", "teams", "computer-use", "acp", "mistral", "bedrock", "vertex", "azure-identity", "termux", "termux-all", "dingtalk", "feishu", "google", "youtube", "web", "all"]
 
 [[package]]
 name = "hf-xet"
diff --git a/website/docs/integrations/index.md b/website/docs/integrations/index.md
index 7f4e1f17679..7f875aaa080 100644
--- a/website/docs/integrations/index.md
+++ b/website/docs/integrations/index.md
@@ -91,7 +91,7 @@ Speech-to-text supports six providers: local faster-whisper (free, runs on-devic
 
 Hermes runs as a gateway bot on 27+ messaging platforms, all configured through the same `gateway` subsystem:
 
-- **[Telegram](/user-guide/messaging/telegram)**, **[Discord](/user-guide/messaging/discord)**, **[Slack](/user-guide/messaging/slack)**, **[WhatsApp](/user-guide/messaging/whatsapp)**, **[Signal](/user-guide/messaging/signal)**, **[Matrix](/user-guide/messaging/matrix)**, **[Mattermost](/user-guide/messaging/mattermost)**, **[Email](/user-guide/messaging/email)**, **[SMS](/user-guide/messaging/sms)**, **[DingTalk](/user-guide/messaging/dingtalk)**, **[Feishu/Lark](/user-guide/messaging/feishu)**, **[WeCom](/user-guide/messaging/wecom)**, **[WeCom Callback](/user-guide/messaging/wecom-callback)**, **[Weixin](/user-guide/messaging/weixin)**, **[BlueBubbles](/user-guide/messaging/bluebubbles)**, **[QQ Bot](/user-guide/messaging/qqbot)**, **[Yuanbao](/user-guide/messaging/yuanbao)**, **[Home Assistant](/user-guide/messaging/homeassistant)**, **[Microsoft Teams](/user-guide/messaging/teams)**, **[Microsoft Teams Meetings](/user-guide/messaging/teams-meetings)**, **[Microsoft Graph Webhook](/user-guide/messaging/msgraph-webhook)**, **[Google Chat](/user-guide/messaging/google_chat)**, **[LINE](/user-guide/messaging/line)**, **[ntfy](/user-guide/messaging/ntfy)**, **[SimpleX](/user-guide/messaging/simplex)**, **[Open WebUI](/user-guide/messaging/open-webui)**, **[Webhooks](/user-guide/messaging/webhooks)**
+- **[Telegram](/user-guide/messaging/telegram)**, **[Discord](/user-guide/messaging/discord)**, **[Slack](/user-guide/messaging/slack)**, **[WhatsApp](/user-guide/messaging/whatsapp)**, **[Signal](/user-guide/messaging/signal)**, **[Matrix](/user-guide/messaging/matrix)**, **[Mattermost](/user-guide/messaging/mattermost)**, **[Email](/user-guide/messaging/email)**, **[SMS](/user-guide/messaging/sms)**, **[DingTalk](/user-guide/messaging/dingtalk)**, **[Feishu/Lark](/user-guide/messaging/feishu)**, **[WeCom](/user-guide/messaging/wecom)**, **[WeCom Callback](/user-guide/messaging/wecom-callback)**, **[Weixin](/user-guide/messaging/weixin)**, **[BlueBubbles](/user-guide/messaging/bluebubbles)**, **[QQ Bot](/user-guide/messaging/qqbot)**, **[Yuanbao](/user-guide/messaging/yuanbao)**, **[Home Assistant](/user-guide/messaging/homeassistant)**, **[Microsoft Teams](/user-guide/messaging/teams)**, **[Microsoft Teams Meetings](/user-guide/messaging/teams-meetings)**, **[Microsoft Graph Webhook](/user-guide/messaging/msgraph-webhook)**, **[Google Chat](/user-guide/messaging/google_chat)**, **[LINE](/user-guide/messaging/line)**, **[ntfy](/user-guide/messaging/ntfy)**, **[SimpleX](/user-guide/messaging/simplex)**, **[X Chat](/user-guide/messaging/xchat)**, **[Open WebUI](/user-guide/messaging/open-webui)**, **[Webhooks](/user-guide/messaging/webhooks)**
 
 See the [Messaging Gateway overview](/user-guide/messaging) for the platform comparison table and setup guide.
 
diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md
index bb9f22882e8..52b79d7c86a 100644
--- a/website/docs/reference/environment-variables.md
+++ b/website/docs/reference/environment-variables.md
@@ -621,6 +621,28 @@ Connect Hermes to a [SimpleX Chat](https://simplex.chat/) network via a local `s
 | `SIMPLEX_HOME_CHANNEL` | Default contact/group ID for cron / notification delivery. |
 | `SIMPLEX_HOME_CHANNEL_NAME` | Human label for the home channel (defaults to the ID). |
 
+### X Chat
+
+Connect Hermes to [X Chat](https://docs.x.com/xchat/introduction) — X's end-to-end encrypted direct messages. See [the X Chat messaging guide](/user-guide/messaging/xchat).
+
+| Variable | Description |
+|----------|-------------|
+| `XCHAT_ACCESS_TOKEN` | OAuth2 user access token with `dm.read`, `dm.write`, `users.read`, `tweet.read` scopes (set by `hermes xchat setup`). |
+| `XCHAT_REFRESH_TOKEN` | OAuth2 refresh token — enables automatic renewal of the ~2h access token (rotated on every refresh and re-persisted). |
+| `XCHAT_CLIENT_ID` | X app OAuth2 client id — required for automatic token refresh. |
+| `XCHAT_CLIENT_SECRET` | X app OAuth2 client secret (confidential clients only). |
+| `XCHAT_USER_ID` | The bot account's numeric X user id (derived automatically by `hermes xchat setup`). |
+| `XCHAT_SIGNING_KEY_VERSION` | Registered public-key version for message signing (written by `hermes xchat setup`). |
+| `XCHAT_PRIVATE_KEYS_B64` | Base64 private-key blob override — takes precedence over `~/.hermes/xchat/private_keys.b64`. |
+| `XCHAT_ALLOWED_USERS` | Comma-separated numeric X user ids allowed to talk to the bot. |
+| `XCHAT_ALLOW_ALL_USERS` | Allow any sender to trigger the bot (dev only — disables allowlist). |
+| `XCHAT_CONVERSATION_IDS` | Comma-separated conversation ids to poll (omit to auto-discover all conversations). |
+| `XCHAT_POLL_INTERVAL` | Seconds between event polls (default `10`, floor `2`). |
+| `XCHAT_REQUIRE_MENTION` | Ignore group-conversation messages unless they match a mention wake word (`true`/`false`, default `false`). |
+| `XCHAT_MENTION_PATTERNS` | Mention wake-word regexes for group chats (JSON list or comma/newline-separated; defaults to the Hermes wake words). |
+| `XCHAT_HOME_CHANNEL` | Default conversation/user id for cron / notification delivery. |
+| `XCHAT_HOME_CHANNEL_NAME` | Human label for the home channel (defaults to the id). |
+
 ### Photon
 
 Connect Hermes to [Photon](https://photon.codes/) / Spectrum (iMessage and other Spectrum platforms) via the Node sidecar. See [the Photon messaging guide](/user-guide/messaging/photon).
diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md
index 88f57010064..768c2e1e9f9 100644
--- a/website/docs/user-guide/messaging/index.md
+++ b/website/docs/user-guide/messaging/index.md
@@ -6,7 +6,7 @@ description: "Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal,
 
 # Messaging Gateway
 
-Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, Weixin, BlueBubbles (iMessage), QQ, Yuanbao, Microsoft Teams, LINE, ntfy, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages.
+Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, X Chat, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, Weixin, BlueBubbles (iMessage), QQ, Yuanbao, Microsoft Teams, LINE, ntfy, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages.
 
 For the full voice feature set — including CLI microphone mode, spoken replies in messaging, and Discord voice-channel conversations — see [Voice Mode](/user-guide/features/voice-mode) and [Use Voice Mode with Hermes](/guides/use-voice-mode-with-hermes).
 
@@ -25,6 +25,7 @@ Bots need both a model provider and tool providers (TTS, web). A [Nous Portal](/
 | WhatsApp | — | ✅ | ✅ | — | — | ✅ | ✅ |
 | Signal | — | ✅ | ✅ | — | — | ✅ | ✅ |
 | SMS | — | — | — | — | — | — | — |
+| X Chat | — | — | — | — | — | ✅ | — |
 | Email | — | ✅ | ✅ | ✅ | — | — | — |
 | Home Assistant | — | — | — | — | — | — | — |
 | Mattermost | ✅ | ✅ | ✅ | ✅ | — | ✅ | ✅ |
@@ -710,6 +711,7 @@ Defaults to `false`. Only platforms whose adapter implements `delete_message` ho
 - [WhatsApp Business Cloud API Setup](whatsapp-cloud.md)
 - [Signal Setup](signal.md)
 - [SMS Setup (Twilio)](sms.md)
+- [X Chat Setup (encrypted X DMs)](xchat.md)
 - [Email Setup](email.md)
 - [Home Assistant Integration](homeassistant.md)
 - [Mattermost Setup](mattermost.md)
diff --git a/website/docs/user-guide/messaging/xchat.md b/website/docs/user-guide/messaging/xchat.md
new file mode 100644
index 00000000000..f19a2eb1ac2
--- /dev/null
+++ b/website/docs/user-guide/messaging/xchat.md
@@ -0,0 +1,101 @@
+# X Chat (encrypted X DMs)
+
+[X Chat](https://docs.x.com/xchat/introduction) is X's end-to-end encrypted direct-message system. The Hermes adapter connects your agent to a bot X account's DMs: message bodies are encrypted and decrypted **locally** with the official Chat XDK — X only ever routes ciphertext, and every message is signed so recipients can verify the sender.
+
+> Run `hermes xchat setup` for a guided walk-through, or pick **X Chat** in `hermes gateway setup`.
+
+## Prerequisites
+
+- An **X developer account** with an app configured for **OAuth 2.0 user context** ([Developer Console](https://developer.x.com/en/portal/dashboard)). X Chat endpoints require API access on your developer plan.
+- A **user access token** for the bot account with scopes: `dm.read`, `dm.write`, `users.read`, `tweet.read` (add `offline.access` to receive a refresh token so Hermes can auto-renew the ~2-hour access token).
+- Python 3.10+ (the `chatxdk` E2EE binding is lazy-installed at first use).
+
+## Setup
+
+```bash
+hermes xchat setup
+```
+
+The wizard:
+
+1. Stores the OAuth2 access token (and optional refresh token + client id) in `~/.hermes/.env`.
+2. Derives the bot account's numeric user id via `GET /2/users/me`.
+3. Generates the E2EE identity + signing keypairs with the Chat XDK, saves the private-key blob to `~/.hermes/xchat/private_keys.b64` (mode 600), and registers the public keys with the X API.
+
+Key registration is **rate limited to a few writes per 24 hours** per account. The setup is resume-safe: the key blob and registration payload are persisted *before* any network call, so an interrupted or rate-limited run resumes the same identity instead of minting a new one.
+
+Check state anytime:
+
+```bash
+hermes xchat status
+```
+
+## Environment variables
+
+| Variable | Required | Description |
+|---|---|---|
+| `XCHAT_ACCESS_TOKEN` | Yes | OAuth2 user access token (dm.read, dm.write, users.read, tweet.read) |
+| `XCHAT_REFRESH_TOKEN` | Recommended | Refresh token — enables automatic access-token renewal (rotated on every refresh and re-persisted) |
+| `XCHAT_CLIENT_ID` | With refresh | X app OAuth2 client id (required for token refresh) |
+| `XCHAT_CLIENT_SECRET` | Optional | Only for confidential clients |
+| `XCHAT_USER_ID` | Auto | Bot account's numeric user id (derived by setup) |
+| `XCHAT_SIGNING_KEY_VERSION` | Auto | Registered public-key version (written by setup) |
+| `XCHAT_PRIVATE_KEYS_B64` | Optional | Key blob override — takes precedence over the blob file |
+| `XCHAT_ALLOWED_USERS` | Recommended | Comma-separated numeric X user ids allowed to talk to the bot |
+| `XCHAT_ALLOW_ALL_USERS` | Optional | `true` allows every sender (dev only) |
+| `XCHAT_CONVERSATION_IDS` | Optional | Pin specific conversation ids to poll; omit to auto-discover |
+| `XCHAT_POLL_INTERVAL` | Optional | Seconds between event polls (default `10`, floor `2`) |
+| `XCHAT_REQUIRE_MENTION` | Optional | In group conversations, only respond when a wake word matches (default `false`) |
+| `XCHAT_MENTION_PATTERNS` | Optional | Custom wake-word regexes (JSON list or comma-separated) |
+| `XCHAT_HOME_CHANNEL` | Optional | Default conversation/user id for cron delivery |
+| `XCHAT_HOME_CHANNEL_NAME` | Optional | Human label for the home channel |
+
+## How it works
+
+- **Inbound** — the adapter polls each conversation's events endpoint. On first sight of a conversation it batch-decrypts the backlog (`decrypt_events`) to seed the SDK's verified conversation-key cache **without replying to old messages**, then decrypts new events individually. `KeyChange` events (conversation-key rotations) are verified and folded into the key cache automatically.
+- **Outbound** — replies are encrypted and signed locally (`encrypt_message` with the session identity), then POSTed as ciphertext.
+- **Senders** — each new sender's public keys are fetched once and pushed into the XDK's signing-key store so their message signatures verify.
+- **Identity** — user ids are numeric X user ids; conversation ids look like `123-456` (1:1) or `g123…` (group).
+
+## Authorization
+
+By default all senders are denied. Either:
+
+1. Set `XCHAT_ALLOWED_USERS` to a comma-separated list of numeric X user ids, or
+2. Use **DM pairing** — an unknown sender gets a pairing code; approve with `hermes pairing approve xchat <CODE>`.
+
+## Group conversations
+
+Group chats (`g…` conversation ids) work out of the box. To keep the bot quiet unless addressed:
+
+```
+XCHAT_REQUIRE_MENTION=true
+```
+
+The default wake words are `hermes` / `hermes agent`; override with `XCHAT_MENTION_PATTERNS`. DMs are never gated.
+
+## Using X Chat with cron jobs
+
+```python
+cronjob(
+    action="create",
+    schedule="every 1h",
+    deliver="xchat",            # uses XCHAT_HOME_CHANNEL
+    prompt="Check for alerts and summarise."
+)
+```
+
+Or target a conversation directly:
+
+```bash
+hermes send xchat:<conversation-id> "Done!"
+```
+
+Standalone delivery opens an ephemeral E2EE session, seeds the conversation key from the conversation's backlog, encrypts, and sends — no running gateway required.
+
+## Limitations
+
+- **Text only for now.** Encrypted media upload/download (the streaming-encrypt flow + `media_hash_key` endpoints) is not wired yet; inbound attachments surface as text-free events and are skipped.
+- **Reply flows only.** The bot answers conversations that exist; initiating a brand-new conversation (which requires a conversation-key handshake) is not supported yet.
+- **Polling latency.** Inbound uses REST polling (default 10s). Webhook / activity-stream delivery may come later.
+- **Access tier.** X Chat API availability depends on your X developer plan.
diff --git a/website/sidebars.ts b/website/sidebars.ts
index 76b948aa725..1928fae495b 100644
--- a/website/sidebars.ts
+++ b/website/sidebars.ts
@@ -641,6 +641,7 @@ const sidebars: SidebarsConfig = {
             'user-guide/messaging/signal',
             'user-guide/messaging/email',
             'user-guide/messaging/sms',
+            'user-guide/messaging/xchat',
           ],
         },
         {