From 46da6784aa9f8950a39931c69112b3aac34344cd Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 12 Aug 2026 17:16:41 -0700
Subject: [PATCH] =?UTF-8?q?feat(xchat):=20full=20feature=20parity=20?=
 =?UTF-8?q?=E2=80=94=20encrypted=20media,=20threaded=20replies,=20new-conv?=
 =?UTF-8?q?ersation=20handshake,=20key-event=20meta,=20read=20receipts?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Brings the X Chat adapter to parity with mature gateway platforms:

- Encrypted media, both directions. Inbound attachments are downloaded
  (GET /2/chat/media/{conv}/{hash}), decrypted with the conversation key
  for the EVENT's key version (post-rotation media stays readable),
  size-capped, cached locally, and surfaced on MessageEvent
  (media_urls/media_types + correct MessageType) so vision/file tools
  see them. Outbound send_image/send_image_file/send_voice/send_video/
  send_document encrypt with the latest conversation key
  (encrypt_stream), upload via the 3-step chat-media flow
  (initialize/append/finalize, base64 JSON segments), and attach by
  media_hash_key. Standalone sends (cron/send_message_tool) carry
  media_files the same way.
- Native threaded replies. A bounded per-conversation cache of decrypted
  events lets send(reply_to=...) use encrypt_reply against the real
  target event; unknown targets fall back to a plain send. Inbound
  reply context (reply_to_message_id/text/author,
  reply_to_is_own_message) now propagates on MessageEvent.
- New-conversation initiation. Standalone send to a bare numeric user id
  performs the conversation-key handshake: fetch both parties' public
  keys, verify each identity↔signing binding (verify_key_binding — a
  substituted key must never receive the conversation key), wrap a
  fresh key per participant (prepare_conversation_key_change), POST to
  add-conversation-keys, then encrypt under the returned raw key.
- meta.conversation_key_events. The events endpoint returns KeyChange
  events SEPARATELY in meta — previously they were never decrypted, so
  conversations whose key changes fell outside the data array could
  never seed a key. Both the poll loop and the standalone sender now
  feed them through the batch decrypt path (after signing-key
  registration) before processing messages.
- Read receipts (opt-in, XCHAT_SEND_READ_RECEIPTS, default off) via
  POST /2/chat/conversations/{id}/read.
- Latest-key-version tracking per conversation for media encrypt and
  correct key selection after rotations.

api.py: media_upload (chunked 3-step), media_download, mark_read.
crypto.py: encrypt_reply, encrypt_media/decrypt_media, verify_key_binding,
prepare_conversation_key_change (SDK->API body mapping incl.
action_signatures), attachments/explicit-key support on encrypt_text,
latest_key_version surfaced from decrypt_events, message_attachments,
detect_mime_type/detect_image_dimensions helpers.

Docs: media/replies/handshake/read-receipts documented; stale "text
only" / "reply flows only" limitations removed; capability row added to
the messaging comparison table; media.write scope noted in setup + docs.

Tests: 51 total — inbound attachment decrypt-and-cache, outbound
encrypt-upload-attach (+ no-key failure), threaded-reply cache hit and
fallback, meta key-event absorption order, read-receipt opt-in/default,
reply-context propagation, full handshake happy path (bindings verified,
key change POSTed, explicit key used), chunked upload reassembly,
media download, mark-read body.
---
 plugins/platforms/xchat/adapter.py            | 555 ++++++++++++++++--
 plugins/platforms/xchat/api.py                |  69 +++
 plugins/platforms/xchat/cli.py                |   3 +-
 plugins/platforms/xchat/crypto.py             | 226 ++++++-
 plugins/platforms/xchat/plugin.yaml           |   4 +
 .../platforms/xchat/test_xchat_adapter.py     | 314 +++++++++-
 .../plugins/platforms/xchat/test_xchat_api.py |  66 +++
 .../docs/reference/environment-variables.md   |   1 +
 website/docs/user-guide/messaging/index.md    |   1 +
 website/docs/user-guide/messaging/xchat.md    |  12 +-
 10 files changed, 1183 insertions(+), 68 deletions(-)

diff --git a/plugins/platforms/xchat/adapter.py b/plugins/platforms/xchat/adapter.py
index d429ba05039f0..1fabcf060a053 100644
--- a/plugins/platforms/xchat/adapter.py
+++ b/plugins/platforms/xchat/adapter.py
@@ -47,7 +47,13 @@ from gateway.platforms.base import (
 )
 
 from .api import HTTPX_AVAILABLE, XChatApi, XChatApiError, XChatRateLimited
-from .crypto import XChatCrypto, message_text
+from .crypto import (
+    XChatCrypto,
+    detect_image_dimensions,
+    detect_mime_type,
+    message_attachments,
+    message_text,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -58,6 +64,11 @@ 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
+# Bounded cache of recent decrypted events per conversation — used to build
+# native threaded replies (encrypt_reply needs the decrypted target event).
+REPLY_CACHE_MAX = 200
+# Cap inbound attachments processed per message.
+MAX_INBOUND_ATTACHMENTS = 5
 
 # Group-chat mention wake words — same defaults as the other Hermes channels
 # so group gating behaves identically everywhere.
@@ -240,6 +251,18 @@ class XChatAdapter(BasePlatformAdapter):
         self._cursors: Dict[str, str] = _load_cursors()
         self._seen_event_ids: Dict[str, float] = {}
         self._conversation_keys: Dict[str, Dict[str, bytes]] = {}
+        # Latest verified key version per conversation — media decrypt must
+        # use the key for the EVENT's version, media encrypt the latest.
+        self._latest_key_version: Dict[str, str] = {}
+        # Recent decrypted events per conversation, keyed by event id — lets
+        # send(reply_to=...) build a native threaded reply.
+        self._event_cache: Dict[str, Dict[str, Dict[str, Any]]] = {}
+        # Read receipts (privacy default: off).
+        env_read = os.getenv("XCHAT_SEND_READ_RECEIPTS")
+        if env_read is not None:
+            self._send_read_receipts = env_read.strip().lower() in {"1", "true", "yes"}
+        else:
+            self._send_read_receipts = bool(extra.get("send_read_receipts", False))
 
         # Signing-key roster (accumulated; the SDK store is replaced wholesale)
         self._signing_keys: List[Dict[str, str]] = []
@@ -421,7 +444,11 @@ class XChatAdapter(BasePlatformAdapter):
         # Page until we reach the cursor (or the feed ends) so a burst of
         # more than one page between polls is never dropped. Newest-first
         # on the wire; hard page cap keeps a pathological feed bounded.
+        # KeyChange events arrive SEPARATELY in meta.conversation_key_events
+        # — they must be decrypted (batch path) before the messages that
+        # were encrypted under them, or no conversation key is available.
         collected: List[Dict[str, Any]] = []
+        key_events_b64: List[str] = []
         token: Optional[str] = None
         reached_cursor = False
         for _ in range(10):
@@ -429,6 +456,10 @@ class XChatAdapter(BasePlatformAdapter):
                 conv_id, max_results=50, pagination_token=token
             )
             raw = page.get("data") or []
+            meta = page.get("meta") or {}
+            for kev in meta.get("conversation_key_events") or []:
+                if kev and kev not in key_events_b64:
+                    key_events_b64.append(kev)
             if not raw:
                 break
             for item in raw:
@@ -437,16 +468,24 @@ class XChatAdapter(BasePlatformAdapter):
                     reached_cursor = True
                     break
                 collected.append(item)
-            token = (page.get("meta") or {}).get("next_token")
+            token = meta.get("next_token")
             if reached_cursor or not token:
                 break
-        if not collected:
+        if not collected and not key_events_b64:
             return
 
         # Process oldest-first.
         collected.reverse()
         await self._register_signing_keys(collected)
 
+        # Verify + cache any conversation-key changes FIRST (after signing
+        # keys are registered — an unverifiable KeyChange is dropped by the
+        # SDK), so this poll's messages can decrypt under rotated keys.
+        if key_events_b64:
+            self._absorb_key_batch(conv_id, key_events_b64)
+        if not collected:
+            return
+
         if not cursor:
             # First sight of this conversation EVER (no persisted cursor):
             # batch-decrypt to seed the SDK's verified-key cache, but do NOT
@@ -455,12 +494,7 @@ class XChatAdapter(BasePlatformAdapter):
             # arrived while we were down are processed normally above.
             events_b64 = [e["encoded_event"] for e in collected 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)
+                self._absorb_key_batch(conv_id, events_b64, label="backlog")
             newest = str(collected[-1].get("id") or "")
             if newest:
                 self._set_cursor(conv_id, newest)
@@ -494,12 +528,7 @@ class XChatAdapter(BasePlatformAdapter):
             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)
+                self._absorb_key_batch(conv_id, [event_b64], label="key-change")
                 self._set_cursor(conv_id, event_id)
                 continue
             if etype not in ("Message", "MessageEdit"):
@@ -512,22 +541,155 @@ class XChatAdapter(BasePlatformAdapter):
                 self._set_cursor(conv_id, event_id)
                 continue  # echo of our own reply
 
-            text = message_text(event)
-            if not text:
-                self._set_cursor(conv_id, event_id)
-                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)
+            self._cache_event(canonical_conv, event_id, event)
+
+            # Encrypted media attachments: download + decrypt + cache locally
+            # so vision / the agent can read them.
+            media_urls, media_types = await self._fetch_attachments(
+                conv_id, event, item
+            )
+
+            text = message_text(event) or ""
+            if not text and not media_urls:
+                self._set_cursor(conv_id, event_id)
+                continue
+
             await self._dispatch_inbound(
                 conv_id=canonical_conv,
                 sender_id=sender_id,
                 text=text,
                 message_id=event_id,
                 raw=item,
+                media_urls=media_urls,
+                media_types=media_types,
+                reply_to=self._reply_context(canonical_conv, event),
             )
             self._set_cursor(conv_id, event_id)
+            await self._maybe_mark_read(conv_id, item)
+
+    def _absorb_key_batch(
+        self, conv_id: str, events_b64: List[str], *, label: str = "key-events"
+    ) -> None:
+        """Batch-decrypt events to extract + cache verified conversation keys."""
+        assert self._crypto is not None
+        try:
+            batch = self._crypto.decrypt_batch(events_b64)
+        except Exception as e:
+            logger.warning("[xchat] %s decrypt failed conv=%s: %s", label, conv_id, e)
+            return
+        keys = (batch.get("conversation_keys") or {}).get("keys") or {}
+        if keys:
+            self._conversation_keys.setdefault(conv_id, {}).update(keys)
+        latest = batch.get("latest_key_version")
+        if latest:
+            self._latest_key_version[conv_id] = str(latest)
+
+    def _cache_event(self, conv_id: str, event_id: str, event: Dict[str, Any]) -> None:
+        """Keep a bounded cache of decrypted events for native threaded replies."""
+        cache = self._event_cache.setdefault(conv_id, {})
+        cache[event_id] = event
+        while len(cache) > REPLY_CACHE_MAX:
+            cache.pop(next(iter(cache)))
+
+    def _reply_context(
+        self, conv_id: str, event: Dict[str, Any]
+    ) -> Optional[Dict[str, Any]]:
+        """Reply metadata for the inbound event (id/text/author), if present."""
+        reply = event.get("reply_to") or event.get("replied_to_event")
+        if not reply:
+            return None
+        reply = reply if isinstance(reply, dict) else {}
+        rid = str(reply.get("id") or reply.get("sequence_id") or "") or None
+        return {
+            "message_id": rid,
+            "text": reply.get("text") or message_text(reply),
+            "author_id": str(reply.get("sender_id") or "") or None,
+        }
+
+    async def _fetch_attachments(
+        self, conv_id: str, event: Dict[str, Any], item: Dict[str, Any]
+    ) -> tuple[List[str], List[str]]:
+        """Download + decrypt inbound media attachments to the local cache.
+
+        Uses the conversation key for the EVENT's key version — after a
+        rotation the latest key cannot decrypt older media.
+        """
+        atts = message_attachments(event)
+        if not atts:
+            return [], []
+        assert self._api is not None and self._crypto is not None
+        from gateway.platforms.base import (
+            cache_audio_from_bytes,
+            cache_document_from_bytes,
+            cache_image_from_bytes,
+            cache_video_from_bytes,
+            validate_inbound_media_size,
+        )
+
+        keys = self._conversation_keys.get(conv_id) or {}
+        event_key_version = str(event.get("key_version") or "")
+        conv_key = keys.get(event_key_version) or (
+            keys.get(self._latest_key_version.get(conv_id, "")) if keys else None
+        )
+        if conv_key is None and keys:
+            # Last resort: any cached key (single-key conversations).
+            conv_key = next(iter(keys.values()))
+        if conv_key is None:
+            logger.warning(
+                "[xchat] attachment skipped conv=%s: no conversation key cached", conv_id
+            )
+            return [], []
+
+        media_urls: List[str] = []
+        media_types: List[str] = []
+        for att in atts[:MAX_INBOUND_ATTACHMENTS]:
+            hash_key = str(att.get("media_hash_key") or "")
+            if not hash_key:
+                continue
+            try:
+                blob = await self._api.media_download(conv_id, hash_key)
+                # Raises ValueError when over the inbound media cap.
+                validate_inbound_media_size(len(blob), media_type="attachment")
+                plaintext = self._crypto.decrypt_media(blob, conv_key)
+            except Exception as e:
+                logger.warning(
+                    "[xchat] attachment fetch/decrypt failed conv=%s key=%s: %s",
+                    conv_id, hash_key[:12], e,
+                )
+                continue
+            mime = detect_mime_type(plaintext) or "application/octet-stream"
+            try:
+                if mime.startswith("image/"):
+                    ext = "." + (mime.split("/", 1)[1] or "jpg").replace("jpeg", "jpg")
+                    path = cache_image_from_bytes(plaintext, ext=ext)
+                elif mime.startswith("audio/"):
+                    path = cache_audio_from_bytes(plaintext)
+                elif mime.startswith("video/"):
+                    path = cache_video_from_bytes(plaintext)
+                else:
+                    filename = str(att.get("filename") or f"xchat-{hash_key[:10]}.bin")
+                    path = cache_document_from_bytes(plaintext, filename)
+            except Exception as e:
+                logger.warning("[xchat] attachment cache failed conv=%s: %s", conv_id, e)
+                continue
+            media_urls.append(path)
+            media_types.append(mime)
+        return media_urls, media_types
+
+    async def _maybe_mark_read(self, conv_id: str, item: Dict[str, Any]) -> None:
+        """Best-effort read receipt (opt-in via XCHAT_SEND_READ_RECEIPTS)."""
+        if not self._send_read_receipts or self._api is None:
+            return
+        seq = str(item.get("sequence_id") or item.get("id") or "")
+        if not seq:
+            return
+        try:
+            await self._api.mark_read(conv_id, seq)
+        except Exception:
+            logger.debug("[xchat] mark-read failed conv=%s", conv_id, exc_info=True)
 
     def _set_cursor(self, conv_id: str, event_id: str) -> None:
         """Advance + persist the per-conversation cursor (monotonic)."""
@@ -601,6 +763,9 @@ class XChatAdapter(BasePlatformAdapter):
         text: str,
         message_id: str,
         raw: Dict[str, Any],
+        media_urls: Optional[List[str]] = None,
+        media_types: Optional[List[str]] = None,
+        reply_to: Optional[Dict[str, Any]] = None,
     ) -> None:
         is_group = conv_id.startswith("g")
         chat_type = "group" if is_group else "dm"
@@ -609,7 +774,7 @@ class XChatAdapter(BasePlatformAdapter):
             if not self._message_matches_mention_patterns(text):
                 return
             text = self._clean_mention_text(text)
-            if not text:
+            if not text and not media_urls:
                 return
 
         source = self.build_source(
@@ -620,12 +785,33 @@ class XChatAdapter(BasePlatformAdapter):
             user_name=None,
             message_id=message_id,
         )
+        mtype = MessageType.TEXT
+        if media_urls:
+            first = (media_types or [""])[0]
+            if first.startswith("image/"):
+                mtype = MessageType.PHOTO
+            elif first.startswith("audio/"):
+                mtype = MessageType.VOICE
+            elif first.startswith("video/"):
+                mtype = MessageType.VIDEO
+            else:
+                mtype = MessageType.DOCUMENT
+        reply_to = reply_to or {}
         event = MessageEvent(
             text=text,
-            message_type=MessageType.TEXT,
+            message_type=mtype,
             source=source,
             raw_message=raw,
             message_id=message_id,
+            user_id=sender_id,
+            media_urls=list(media_urls or []),
+            media_types=list(media_types or []),
+            reply_to_message_id=reply_to.get("message_id"),
+            reply_to_text=reply_to.get("text"),
+            reply_to_author_id=reply_to.get("author_id"),
+            reply_to_is_own_message=(
+                str(reply_to.get("author_id") or "") == self._bot_user_id
+            ),
         )
         await self.handle_message(event)
 
@@ -643,20 +829,20 @@ class XChatAdapter(BasePlatformAdapter):
         if len(content) > MAX_MESSAGE_LENGTH:
             content = content[:MAX_MESSAGE_LENGTH]
         try:
-            body = self._crypto.encrypt_text(chat_id, content)
+            body = self._encrypt_outbound(chat_id, content, reply_to=reply_to)
         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).
+            # conversation the bot initiates needs a key handshake — see
+            # initiate_conversation() (used by the standalone sender).
             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."
+                    "always have it. To message a brand-new user, use "
+                    "`hermes send xchat:<user-id>` (it performs the key "
+                    "handshake automatically)."
                 ),
             )
         except Exception as e:
@@ -675,6 +861,162 @@ class XChatAdapter(BasePlatformAdapter):
                 self._seen_event_ids[str(eid)] = time.time()
         return SendResult(success=True, message_id=msg_id)
 
+    def _encrypt_outbound(
+        self,
+        chat_id: str,
+        content: str,
+        *,
+        reply_to: Optional[str] = None,
+        attachments: Optional[List[Dict[str, Any]]] = None,
+    ) -> Dict[str, str]:
+        """Encrypt text (or media caption) — native threaded reply when the
+        replied-to event is in the decrypted-event cache."""
+        assert self._crypto is not None
+        if reply_to:
+            target = (self._event_cache.get(chat_id) or {}).get(str(reply_to))
+            if target is not None:
+                try:
+                    return self._crypto.encrypt_reply(
+                        chat_id, content, target, attachments=attachments
+                    )
+                except Exception:
+                    logger.debug(
+                        "[xchat] encrypt_reply failed conv=%s — plain send", chat_id,
+                        exc_info=True,
+                    )
+        return self._crypto.encrypt_text(chat_id, content, attachments=attachments)
+
+    def _conversation_key_for_send(self, chat_id: str) -> Optional[bytes]:
+        """Latest cached raw conversation key (for media stream encryption)."""
+        keys = self._conversation_keys.get(chat_id) or {}
+        if not keys:
+            return None
+        latest = self._latest_key_version.get(chat_id)
+        if latest and latest in keys:
+            return keys[latest]
+        # Highest numeric version wins when latest is unknown.
+        try:
+            return keys[max(keys, key=lambda v: int(v))]
+        except (ValueError, TypeError):
+            return next(iter(keys.values()))
+
+    async def _send_media_file(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str],
+        reply_to: Optional[str] = None,
+    ) -> SendResult:
+        """Encrypt + upload a file, then send a message carrying the attachment."""
+        if self._api is None or self._crypto is None:
+            return SendResult(success=False, error="xchat adapter not connected")
+        conv_key = self._conversation_key_for_send(chat_id)
+        if conv_key is None:
+            return SendResult(
+                success=False,
+                error="No conversation key cached — cannot encrypt media yet.",
+            )
+        try:
+            plaintext = Path(file_path).read_bytes()
+        except OSError as e:
+            return SendResult(success=False, error=f"cannot read media file: {e}")
+        try:
+            blob = self._crypto.encrypt_media(plaintext, conv_key)
+            media_hash_key = await self._api.media_upload(chat_id, blob)
+        except (XChatApiError, Exception) as e:
+            logger.warning("[xchat] media upload failed conv=%s: %s", chat_id, e)
+            return SendResult(success=False, error=f"media upload failed: {e}")
+
+        att: Dict[str, Any] = {
+            "attachment_type": "media",
+            "media_hash_key": media_hash_key,
+            "filesize_bytes": len(plaintext),
+            "filename": Path(file_path).name,
+        }
+        dims = detect_image_dimensions(plaintext)
+        att["width"], att["height"] = dims if dims else (0, 0)
+        try:
+            body = self._encrypt_outbound(
+                chat_id, caption or "", reply_to=reply_to, attachments=[att]
+            )
+            out = await self._api.send_message(chat_id, body)
+        except ValueError:
+            return SendResult(success=False, error="No verified conversation key.")
+        except XChatApiError as e:
+            return SendResult(success=False, error=str(e))
+        data = out.get("data") or {}
+        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=str(data.get("message_id") or body.get("message_id") or ""),
+        )
+
+    async def send_image(
+        self,
+        chat_id: str,
+        image_url: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        path = image_url
+        if path.startswith("file://"):
+            from urllib.parse import unquote, urlparse
+
+            path = unquote(urlparse(path).path)
+        if not os.path.isfile(path):
+            # Remote URL — fall back to the base implementation (sends URL text).
+            return await super().send_image(chat_id, image_url, caption, reply_to, metadata)
+        return await self._send_media_file(chat_id, path, caption, reply_to)
+
+    async def send_image_file(
+        self,
+        chat_id: str,
+        image_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._send_media_file(chat_id, image_path, caption, reply_to)
+
+    async def send_voice(
+        self,
+        chat_id: str,
+        audio_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._send_media_file(chat_id, audio_path, caption, reply_to)
+
+    async def send_video(
+        self,
+        chat_id: str,
+        video_path: str,
+        caption: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._send_media_file(chat_id, video_path, caption, reply_to)
+
+    async def send_document(
+        self,
+        chat_id: str,
+        file_path: str,
+        caption: Optional[str] = None,
+        file_name: Optional[str] = None,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ) -> SendResult:
+        return await self._send_media_file(chat_id, file_path, caption, reply_to)
+
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         if self._api is None:
             return
@@ -706,6 +1048,7 @@ def _env_enablement() -> Optional[dict]:
         ("XCHAT_SIGNING_KEY_VERSION", "signing_key_version"),
         ("XCHAT_CONVERSATION_IDS", "conversation_ids"),
         ("XCHAT_POLL_INTERVAL", "poll_interval"),
+        ("XCHAT_SEND_READ_RECEIPTS", "send_read_receipts"),
     ):
         val = os.getenv(env, "").strip()
         if val:
@@ -731,10 +1074,12 @@ async def _standalone_send(
     """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).
+    conversation key from the conversation's event backlog (or performs the
+    conversation-key handshake for a brand-new 1:1 given a bare user id),
+    encrypts, sends, and closes. ``media_files`` are encrypted with the
+    conversation key, uploaded via the 3-step chat-media flow, and attached
+    to the message. ``thread_id`` is accepted for signature parity — X Chat
+    has no thread primitive.
     """
     if not HTTPX_AVAILABLE:
         return {"error": "xchat standalone send: httpx not installed"}
@@ -772,11 +1117,18 @@ async def _standalone_send(
         # Seed the conversation key from the backlog (KeyChange events).
         # KeyChange verification needs the participants' signing keys in the
         # SDK store FIRST — decrypt_batch can't verify (and therefore can't
-        # seed the conversation key) without them.
+        # seed the conversation key) without them. KeyChange events arrive
+        # separately in meta.conversation_key_events.
         page = await api.get_events(chat_id, max_results=50)
         raw_events = page.get("data") or []
-        events_b64 = [e["encoded_event"] for e in raw_events if e.get("encoded_event")]
+        meta = page.get("meta") or {}
+        key_events = [k for k in (meta.get("conversation_key_events") or []) if k]
+        events_b64 = key_events + [
+            e["encoded_event"] for e in raw_events if e.get("encoded_event")
+        ]
         canonical = chat_id
+        raw_key: Optional[bytes] = None
+        raw_key_version: Optional[str] = None
         if events_b64:
             sender_ids = {
                 str(e.get("sender_id"))
@@ -807,18 +1159,85 @@ async def _standalone_send(
                     if conv:
                         canonical = str(conv)
                         break
+                conv_keys = (batch.get("conversation_keys") or {}).get("keys") or {}
+                latest = batch.get("latest_key_version")
+                if conv_keys:
+                    if latest and str(latest) in conv_keys:
+                        raw_key_version = str(latest)
+                    else:
+                        try:
+                            raw_key_version = max(conv_keys, key=lambda v: int(v))
+                        except (ValueError, TypeError):
+                            raw_key_version = next(iter(conv_keys))
+                    raw_key = conv_keys[raw_key_version]
             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"
-                )
+        explicit_key: Optional[bytes] = None
+        explicit_key_version: Optional[str] = None
+
+        # Encrypt + upload media attachments (needs the RAW conversation key).
+        attachments: List[Dict[str, Any]] = []
+        media_errors: List[str] = []
+        for mf in media_files or []:
+            # send_message_tool passes (path, is_voice) tuples; accept bare
+            # strings/paths too for direct callers.
+            if isinstance(mf, (tuple, list)) and mf:
+                mpath = str(mf[0])
+            else:
+                mpath = str(getattr(mf, "path", None) or mf)
+            if raw_key is None:
+                media_errors.append(f"{Path(mpath).name}: no raw conversation key")
+                continue
+            try:
+                plaintext = Path(mpath).read_bytes()
+                blob = crypto.encrypt_media(plaintext, raw_key)
+                media_hash_key = await api.media_upload(canonical, blob)
+            except (OSError, XChatApiError, Exception) as e:
+                media_errors.append(f"{Path(mpath).name}: {e}")
+                continue
+            att: Dict[str, Any] = {
+                "attachment_type": "media",
+                "media_hash_key": media_hash_key,
+                "filesize_bytes": len(plaintext),
+                "filename": Path(mpath).name,
             }
+            dims = detect_image_dimensions(plaintext)
+            att["width"], att["height"] = dims if dims else (0, 0)
+            attachments.append(att)
+        if media_errors:
+            logger.warning("[xchat] standalone media skipped: %s", "; ".join(media_errors))
+
+        try:
+            body = crypto.encrypt_text(
+                canonical, message, attachments=attachments or None
+            )
+        except ValueError:
+            # No verified conversation key. For a bare recipient user id
+            # (brand-new 1:1), perform the conversation-key handshake:
+            # verify both parties' key bindings, wrap a fresh key for each,
+            # and POST it — then encrypt under the raw key directly.
+            if not _is_bare_user_id(chat_id):
+                return {
+                    "error": (
+                        "xchat: no verified conversation key — the target must have "
+                        "an existing conversation with the bot, or pass the "
+                        "recipient's bare user id to start a new one"
+                    )
+                }
+            try:
+                init = await _initiate_conversation(api, crypto, user_id, chat_id)
+            except (XChatApiError, ValueError) as e:
+                return {"error": f"xchat: conversation-key handshake failed: {e}"}
+            canonical = init["conversation_id"] or chat_id
+            explicit_key = init["conversation_key"]
+            explicit_key_version = init["conversation_key_version"]
+            body = crypto.encrypt_text(
+                canonical,
+                message,
+                conversation_key=explicit_key,
+                conversation_key_version=explicit_key_version,
+            )
         out = await api.send_message(canonical, body)
         data = out.get("data") or {}
         return {
@@ -835,6 +1254,53 @@ async def _standalone_send(
         await api.aclose()
 
 
+def _is_bare_user_id(chat_id: str) -> bool:
+    """True for a bare numeric X user id (a 1:1 target with no conversation yet)."""
+    return str(chat_id).isdigit()
+
+
+async def _initiate_conversation(
+    api: "XChatApi", crypto: "XChatCrypto", bot_user_id: str, recipient_id: str
+) -> Dict[str, Any]:
+    """Conversation-key handshake for a brand-new 1:1 conversation.
+
+    Fetches both parties' public keys, verifies each record's
+    identity↔signing binding (a substituted identity key must never receive
+    the conversation key), wraps a fresh conversation key for every
+    participant, and POSTs it to the add-conversation-keys endpoint.
+
+    Returns ``{"conversation_id", "conversation_key", "conversation_key_version"}``.
+    """
+    participants: List[Dict[str, str]] = []
+    for uid in (bot_user_id, recipient_id):
+        records = await api.get_public_keys(uid)
+        if not records:
+            raise ValueError(f"user {uid} has no registered X Chat public keys")
+        rec = records[0]
+        if not crypto.verify_key_binding(
+            str(rec.get("public_key") or ""),
+            str(rec.get("signing_public_key") or ""),
+            str(rec.get("identity_public_key_signature") or ""),
+        ):
+            raise ValueError(f"public-key binding verification failed for user {uid}")
+        participants.append(
+            {
+                "user_id": uid,
+                "public_key": str(rec.get("public_key") or ""),
+                "key_version": str(rec.get("public_key_version") or "1"),
+            }
+        )
+
+    prepared = crypto.prepare_conversation_key_change(participants)
+    resp = await api.add_conversation_keys(recipient_id, prepared["body"])
+    data = resp.get("data") or {}
+    return {
+        "conversation_id": str(data.get("conversation_id") or ""),
+        "conversation_key": prepared["conversation_key"],
+        "conversation_key_version": prepared["conversation_key_version"],
+    }
+
+
 def register(ctx) -> None:
     """Plugin entry point — called by the Hermes plugin loader at startup."""
     from . import cli as _cli
@@ -866,7 +1332,10 @@ def register(ctx) -> None:
             "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."
+            "starting with 'g' are group chats. You can send files, images, "
+            "voice notes, and videos as encrypted attachments via MEDIA:<path> "
+            "tags; inbound attachments are decrypted locally and available "
+            "to your vision/file tools."
         ),
     )
 
diff --git a/plugins/platforms/xchat/api.py b/plugins/platforms/xchat/api.py
index 82abfd9f59c7a..1cfbab5952e07 100644
--- a/plugins/platforms/xchat/api.py
+++ b/plugins/platforms/xchat/api.py
@@ -15,6 +15,7 @@ every use).
 from __future__ import annotations
 
 import asyncio
+import base64
 import logging
 import time
 from typing import Any, Awaitable, Callable, Optional
@@ -279,3 +280,71 @@ class XChatApi:
             f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/keys",
             json_body=body,
         )
+
+    async def mark_read(self, conversation_id: str, seen_until_sequence_id: str) -> None:
+        """POST /2/chat/conversations/{id}/read — read receipt up to a sequence id."""
+        await self._request(
+            "POST",
+            f"/2/chat/conversations/{self._conv_path_id(conversation_id)}/read",
+            json_body={"seen_until_sequence_id": str(seen_until_sequence_id)},
+        )
+
+    # -- media (encrypted attachments) ------------------------------------------
+
+    async def media_upload(
+        self, conversation_id: str, encrypted_blob: bytes, *, chunk_size: int = 1024 * 1024
+    ) -> str:
+        """Three-step encrypted-media upload; returns the ``media_hash_key``.
+
+        initialize → append (base64 JSON segments) → finalize. The size
+        reported to initialize is the ENCRYPTED blob size. Requires the
+        ``media.write`` OAuth scope.
+        """
+        conv = self._conv_path_id(conversation_id)
+        init = await self._request(
+            "POST",
+            "/2/chat/media/upload/initialize",
+            json_body={"conversation_id": conv, "total_bytes": len(encrypted_blob)},
+        )
+        data = init.get("data") or {}
+        session_id = str(data.get("session_id") or "")
+        media_hash_key = str(data.get("media_hash_key") or "")
+        if not session_id or not media_hash_key:
+            raise XChatApiError(500, f"media upload initialize returned no session: {init}")
+
+        for index in range(0, (len(encrypted_blob) + chunk_size - 1) // chunk_size):
+            segment = encrypted_blob[index * chunk_size:(index + 1) * chunk_size]
+            await self._request(
+                "POST",
+                f"/2/chat/media/upload/{session_id}/append",
+                json_body={
+                    "conversation_id": conv,
+                    "media_hash_key": media_hash_key,
+                    "segment_index": index,
+                    "media": base64.b64encode(segment).decode("ascii"),
+                },
+            )
+
+        await self._request(
+            "POST",
+            f"/2/chat/media/upload/{session_id}/finalize",
+            json_body={"conversation_id": conv, "media_hash_key": media_hash_key},
+        )
+        return media_hash_key
+
+    async def media_download(self, conversation_id: str, media_hash_key: str) -> bytes:
+        """GET /2/chat/media/{conversation_id}/{media_hash_key} — encrypted blob."""
+        await self.ensure_token()
+        resp = await self._http().get(
+            f"{self._base_url}/2/chat/media/{self._conv_path_id(conversation_id)}/{media_hash_key}",
+            headers={"Authorization": f"Bearer {self._access_token}"},
+        )
+        if resp.status_code == 401 and self.can_refresh:
+            await self._refresh_access_token()
+            resp = await self._http().get(
+                f"{self._base_url}/2/chat/media/{self._conv_path_id(conversation_id)}/{media_hash_key}",
+                headers={"Authorization": f"Bearer {self._access_token}"},
+            )
+        if resp.status_code >= 300:
+            raise XChatApiError(resp.status_code, resp.text[:300])
+        return resp.content
diff --git a/plugins/platforms/xchat/cli.py b/plugins/platforms/xchat/cli.py
index 3115a325466d3..8d80498098c9a 100644
--- a/plugins/platforms/xchat/cli.py
+++ b/plugins/platforms/xchat/cli.py
@@ -178,7 +178,8 @@ def cmd_setup(*, force: bool) -> int:
     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"
+        "(offline.access too if you want refresh tokens; media.write to send\n"
+        "encrypted attachments).\n"
         "Docs: https://docs.x.com/xchat/getting-started\n"
     )
 
diff --git a/plugins/platforms/xchat/crypto.py b/plugins/platforms/xchat/crypto.py
index d8a69992cbcce..1758e9fe3da01 100644
--- a/plugins/platforms/xchat/crypto.py
+++ b/plugins/platforms/xchat/crypto.py
@@ -12,9 +12,16 @@ Responsibilities:
                            :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`
+* message encryption    -> :meth:`XChatCrypto.encrypt_text` (with optional
+                           attachments) and :meth:`XChatCrypto.encrypt_reply`
 * event decryption      -> :meth:`XChatCrypto.decrypt_batch` (decrypt_events)
                            and :meth:`XChatCrypto.decrypt_one` (decrypt_event)
+* media encryption      -> :meth:`XChatCrypto.encrypt_media` /
+                           :meth:`XChatCrypto.decrypt_media` (stream cipher
+                           under the conversation key)
+* conversation keys     -> :meth:`XChatCrypto.prepare_conversation_key_change`
+                           + :meth:`XChatCrypto.verify_key_binding` (initiate
+                           brand-new conversations / rotate keys)
 
 The decrypted-event dict shape follows the Chat XDK: ``{"type": "Message",
 "id": ..., "sender_id": ..., "conversation_id": ..., "content": {"text":
@@ -42,19 +49,49 @@ def _as_dict(obj: Any) -> dict[str, Any]:
         return {}
 
 
-def _load_chat_class():
-    """Import (lazy-installing if needed) and return ``chat_xdk.Chat``."""
+def _load_chat_xdk():
+    """Import (lazy-installing if needed) and return the ``chat_xdk`` module."""
     try:
-        from chat_xdk import Chat  # type: ignore[import-not-found]
-        return Chat
+        import chat_xdk  # type: ignore[import-not-found]
+        return chat_xdk
     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
+    import chat_xdk  # type: ignore[import-not-found]
+    return chat_xdk
+
+
+def _load_chat_class():
+    """Import (lazy-installing if needed) and return ``chat_xdk.Chat``."""
+    return _load_chat_xdk().Chat
+
+
+def detect_mime_type(data: bytes) -> Optional[str]:
+    """MIME sniff on PLAINTEXT bytes (Chat XDK helper)."""
+    try:
+        return _load_chat_xdk().detect_mime_type(bytes(data))
+    except Exception:
+        return None
+
+
+def detect_image_dimensions(data: bytes) -> Optional[tuple[int, int]]:
+    """(width, height) of PLAINTEXT image bytes, or None (Chat XDK helper)."""
+    try:
+        dims = _load_chat_xdk().detect_image_dimensions(bytes(data))
+    except Exception:
+        return None
+    if not dims:
+        return None
+    try:
+        # Bindings return either a (w, h) tuple or an object with attributes.
+        if isinstance(dims, (tuple, list)):
+            return int(dims[0]), int(dims[1])
+        return int(dims.width), int(dims.height)
+    except Exception:
+        return None
 
 
 class XChatCrypto:
@@ -115,6 +152,80 @@ class XChatCrypto:
         blob_b64 = base64.b64encode(bytes(exported)).decode("ascii") if exported else ""
         return {"registration": body, "version": version, "private_keys_b64": blob_b64}
 
+    def verify_key_binding(
+        self,
+        identity_public_key_b64: str,
+        signing_public_key_b64: str,
+        identity_public_key_signature_b64: str,
+    ) -> bool:
+        """Verify a fetched public-key record's identity↔signing binding.
+
+        MUST be called on every record before wrapping a conversation key to
+        it (``prepare_conversation_key_change`` encrypts to whatever you pass
+        — a substituted identity key would silently receive the key).
+        """
+        try:
+            return bool(
+                self.chat.verify_key_binding(
+                    identity_public_key_b64,
+                    signing_public_key_b64,
+                    identity_public_key_signature_b64,
+                )
+            )
+        except Exception:
+            return False
+
+    # -- Conversation-key setup (initiation / rotation) -----------------------
+
+    def prepare_conversation_key_change(
+        self,
+        public_keys: list[dict[str, str]],
+        *,
+        conversation_id: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """Generate + wrap a fresh conversation key for every participant.
+
+        ``public_keys``: ``[{"user_id", "public_key", "key_version"}, ...]``
+        (verified via :meth:`verify_key_binding` first). Returns the API body
+        for ``POST /2/chat/conversations/{id}/keys`` plus the raw key:
+
+        ``{"body": {...}, "conversation_key": bytes,
+          "conversation_key_version": str}``
+        """
+        prepared = _as_dict(
+            self.chat.prepare_conversation_key_change(
+                public_keys, conversation_id=conversation_id
+            )
+        )
+        body = {
+            "conversation_key_version": prepared["conversation_key_version"],
+            "conversation_participant_keys": [
+                {
+                    "user_id": pk["user_id"],
+                    "encrypted_conversation_key": pk["encrypted_key"],
+                    "public_key_version": pk["public_key_version"],
+                }
+                for pk in (prepared.get("participant_keys") or [])
+            ],
+            "action_signatures": [
+                {
+                    "message_id": sig["message_id"],
+                    "encoded_message_event_detail": sig["encoded_message_event_detail"],
+                    "message_event_signature": {
+                        "signature": sig["signature"],
+                        "public_key_version": sig["public_key_version"],
+                        "signature_version": sig["signature_version"],
+                    },
+                }
+                for sig in (prepared.get("action_signatures") or [])
+            ],
+        }
+        return {
+            "body": body,
+            "conversation_key": prepared.get("conversation_key"),
+            "conversation_key_version": str(prepared["conversation_key_version"]),
+        }
+
     # -- Decryption ----------------------------------------------------------
 
     def decrypt_batch(self, events_b64: list[str]) -> dict[str, Any]:
@@ -124,15 +235,25 @@ class XChatCrypto:
         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.
+
+        NOTE: the events endpoint returns KeyChange events SEPARATELY in
+        ``meta.conversation_key_events`` — callers must prepend those to the
+        batch or no conversation key is ever extracted.
         """
         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 [])
         ]
+        conv_keys = result.get("conversation_keys") or {}
         return {
             "messages": messages,
-            "conversation_keys": result.get("conversation_keys") or {},
+            "conversation_keys": conv_keys,
+            "latest_key_version": (
+                str(conv_keys.get("latest_version"))
+                if conv_keys.get("latest_version") is not None
+                else None
+            ),
             "errors": result.get("errors") or {},
         }
 
@@ -144,20 +265,77 @@ class XChatCrypto:
 
     # -- Encryption ----------------------------------------------------------
 
-    def encrypt_text(self, conversation_id: str, text: str) -> dict[str, str]:
-        """Encrypt + sign ``text``, returning the X API send-message body.
+    def encrypt_text(
+        self,
+        conversation_id: str,
+        text: str,
+        *,
+        attachments: Optional[list[dict[str, Any]]] = None,
+        conversation_key: Optional[bytes] = None,
+        conversation_key_version: Optional[str] = None,
+    ) -> dict[str, str]:
+        """Encrypt + sign ``text`` (optionally with media attachments).
 
-        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.
+        Returns the X API send-message body. The conversation key is resolved
+        from the SDK's verified-key cache (``set_cache_keys``) unless an
+        explicit ``conversation_key`` + version pair is given (used right
+        after key initiation, before any KeyChange event has been polled).
+        Raises ``ValueError`` when no verified key is available.
         """
-        payload = self.chat.encrypt_message(str(conversation_id), text)
+        kwargs: dict[str, Any] = {}
+        if attachments:
+            kwargs["attachments"] = attachments
+        if conversation_key is not None:
+            kwargs["conversation_key"] = conversation_key
+            kwargs["conversation_key_version"] = conversation_key_version
+        payload = self.chat.encrypt_message(str(conversation_id), text, **kwargs)
         return {
             "message_id": payload.message_id,
             "encoded_message_create_event": payload.encrypted_content,
             "encoded_message_event_signature": payload.encoded_event_signature,
         }
 
+    def encrypt_reply(
+        self,
+        conversation_id: str,
+        text: str,
+        reply_to_event: dict[str, Any],
+        *,
+        attachments: Optional[list[dict[str, Any]]] = None,
+    ) -> dict[str, str]:
+        """Encrypt + sign ``text`` as a native threaded reply.
+
+        ``reply_to_event`` is the DECRYPTED event dict of the message being
+        replied to (from :meth:`decrypt_one` / :meth:`decrypt_batch`).
+        Falls back to :meth:`encrypt_text` semantics on SDK versions without
+        reply support.
+        """
+        kwargs: dict[str, Any] = {}
+        if attachments:
+            kwargs["attachments"] = attachments
+        payload = self.chat.encrypt_reply(
+            str(conversation_id), text, reply_to_event=reply_to_event, **kwargs
+        )
+        return {
+            "message_id": payload.message_id,
+            "encoded_message_create_event": payload.encrypted_content,
+            "encoded_message_event_signature": payload.encoded_event_signature,
+        }
+
+    # -- Media (stream cipher under the conversation key) ---------------------
+
+    def encrypt_media(self, plaintext: bytes, conversation_key: bytes) -> bytes:
+        """Encrypt attachment bytes for upload (whole payload in memory)."""
+        return bytes(self.chat.encrypt_stream(bytes(plaintext), conversation_key))
+
+    def decrypt_media(self, ciphertext: bytes, conversation_key: bytes) -> bytes:
+        """Decrypt a downloaded attachment blob.
+
+        The key MUST be the conversation key for the *event's* key version —
+        after a rotation, the latest key cannot decrypt older media.
+        """
+        return bytes(self.chat.decrypt_stream(bytes(ciphertext), conversation_key))
+
 
 def message_text(event: dict[str, Any]) -> Optional[str]:
     """Pull the plain text out of a decrypted Message/MessageEdit event.
@@ -172,3 +350,23 @@ def message_text(event: dict[str, Any]) -> Optional[str]:
     if isinstance(content, dict):
         return content.get("text")
     return None
+
+
+def message_attachments(event: dict[str, Any]) -> list[dict[str, Any]]:
+    """Media attachment descriptors from a decrypted Message/MessageEdit event.
+
+    Each entry carries ``media_hash_key`` (+ optional filename/width/height/
+    filesize_bytes). Returns [] for non-message events or text-only messages.
+    """
+    if event.get("type") not in ("Message", "MessageEdit"):
+        return []
+    content = event.get("content") or {}
+    if not isinstance(content, dict):
+        return []
+    raw = content.get("attachments") or event.get("attachments") or []
+    out: list[dict[str, Any]] = []
+    for att in raw if isinstance(raw, (list, tuple)) else []:
+        att = _as_dict(att)
+        if att.get("media_hash_key"):
+            out.append(att)
+    return out
diff --git a/plugins/platforms/xchat/plugin.yaml b/plugins/platforms/xchat/plugin.yaml
index f471858c90604..3c0f94916659a 100644
--- a/plugins/platforms/xchat/plugin.yaml
+++ b/plugins/platforms/xchat/plugin.yaml
@@ -62,6 +62,10 @@ optional_env:
     description: "Seconds between event polls per cycle (default 10)"
     prompt: "Poll interval seconds (default 10)"
     password: false
+  - name: XCHAT_SEND_READ_RECEIPTS
+    description: "Send read receipts for processed messages (true/false, default false)"
+    prompt: "Send read receipts? (true/false)"
+    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?"
diff --git a/tests/plugins/platforms/xchat/test_xchat_adapter.py b/tests/plugins/platforms/xchat/test_xchat_adapter.py
index ebb834994391f..e754df7d8396a 100644
--- a/tests/plugins/platforms/xchat/test_xchat_adapter.py
+++ b/tests/plugins/platforms/xchat/test_xchat_adapter.py
@@ -37,24 +37,51 @@ class FakeCrypto:
         self.signing_keys: List[Dict[str, str]] = []
         self.decrypt_map: Dict[str, Dict[str, Any]] = {}
         self.fail_encrypt: Optional[Exception] = None
+        self.replies: List[tuple] = []
+        self.media_encrypted: List[bytes] = []
+        self.media_decrypted: List[bytes] = []
 
     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": {}}
+        return {
+            "messages": [],
+            "conversation_keys": {"keys": {"1": b"k"}},
+            "latest_key_version": "1",
+            "errors": {},
+        }
 
-    def encrypt_text(self, conversation_id, text):
+    def encrypt_text(self, conversation_id, text, *, attachments=None, **kw):
         if self.fail_encrypt is not None:
             raise self.fail_encrypt
-        self.encrypted.append((conversation_id, text))
+        self.encrypted.append((conversation_id, text, attachments))
         return {
             "message_id": "mid-1",
             "encoded_message_create_event": "ZW5j",
             "encoded_message_event_signature": "c2ln",
         }
 
+    def encrypt_reply(self, conversation_id, text, reply_to_event, *, attachments=None):
+        if self.fail_encrypt is not None:
+            raise self.fail_encrypt
+        self.replies.append((conversation_id, text, reply_to_event, attachments))
+        return {
+            "message_id": "mid-r",
+            "encoded_message_create_event": "cmVw",
+            "encoded_message_event_signature": "c2ln",
+        }
+
+    def encrypt_media(self, plaintext, conversation_key):
+        self.media_encrypted.append(bytes(plaintext))
+        return b"ENC" + bytes(plaintext)
+
+    def decrypt_media(self, ciphertext, conversation_key):
+        self.media_decrypted.append(bytes(ciphertext))
+        assert bytes(ciphertext).startswith(b"ENC")
+        return bytes(ciphertext)[3:]
+
     def set_signing_keys(self, keys):
         self.signing_keys = list(keys)
 
@@ -65,6 +92,10 @@ class FakeApi:
     def __init__(self) -> None:
         self.sent: List[tuple] = []
         self.typing: List[str] = []
+        self.reads: List[tuple] = []
+        self.uploads: List[tuple] = []
+        self.key_changes: List[tuple] = []
+        self.media_blobs: Dict[str, bytes] = {}
         self.public_keys: Dict[str, List[Dict[str, Any]]] = {}
         self.events_pages: Dict[str, Dict[str, Any]] = {}
         # Optional multi-page feed: {conv_id: {pagination_token_or_None: page}}
@@ -97,6 +128,20 @@ class FakeApi:
     async def send_typing(self, conversation_id):
         self.typing.append(conversation_id)
 
+    async def mark_read(self, conversation_id, seen_until_sequence_id):
+        self.reads.append((conversation_id, seen_until_sequence_id))
+
+    async def media_upload(self, conversation_id, encrypted_blob, *, chunk_size=1024 * 1024):
+        self.uploads.append((conversation_id, bytes(encrypted_blob)))
+        return f"mhk-{len(self.uploads)}"
+
+    async def media_download(self, conversation_id, media_hash_key):
+        return self.media_blobs[media_hash_key]
+
+    async def add_conversation_keys(self, conversation_id, body):
+        self.key_changes.append((conversation_id, body))
+        return {"data": {"conversation_id": "111:999", "sequence_id": "sq1"}}
+
     async def aclose(self):
         pass
 
@@ -496,7 +541,7 @@ async def test_send_encrypts_and_posts(monkeypatch):
     result = await adapter.send("111:999", "hi there")
     assert result.success
     assert result.message_id == "mid-1"
-    assert crypto.encrypted == [("111:999", "hi there")]
+    assert crypto.encrypted == [("111:999", "hi there", None)]
     conv, body = api.sent[0]
     assert conv == "111:999"
     assert body["encoded_message_create_event"] == "ZW5j"
@@ -685,7 +730,7 @@ class StrictCrypto:
             "errors": {},
         }
 
-    def encrypt_text(self, conversation_id, text):
+    def encrypt_text(self, conversation_id, text, *, attachments=None, **kw):
         if not self.keys_seeded:
             raise ValueError("no verified conversation key")
         self.encrypted.append((conversation_id, text))
@@ -753,3 +798,262 @@ async def test_standalone_send_no_key_without_signing_roster(monkeypatch):
     cfg = PlatformConfig(enabled=True, extra={})
     out = await xchat_adapter._standalone_send(cfg, "111-999", "hello")
     assert "no verified conversation key" in out.get("error", "")
+
+
+# ---------------------------------------------------------------------------
+# Media, replies, read receipts, key-events meta
+
+
+@pytest.mark.asyncio
+async def test_inbound_attachment_downloaded_and_decrypted(monkeypatch, tmp_path):
+    """Encrypted inbound media is downloaded, decrypted with the event's
+    key version, cached locally, and surfaced on the MessageEvent."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._cursors["111-999"] = "e1"
+    adapter._conversation_keys["111-999"] = {"1": b"k"}
+    adapter._latest_key_version["111-999"] = "1"
+
+    # PNG magic so mime sniffing (real chatxdk helper unavailable → fallback)
+    png = b"\x89PNG\r\n\x1a\n" + b"0" * 32
+    api.media_blobs["mhk-img"] = b"ENC" + png
+    crypto.decrypt_map["MED"] = {
+        "type": "Message",
+        "id": "e9",
+        "sender_id": "111",
+        "conversation_id": "111:999",
+        "key_version": "1",
+        "content": {
+            "text": "look at this",
+            "attachments": [{"media_hash_key": "mhk-img", "filename": "photo.png"}],
+        },
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e9", "encoded_event": "MED", "sender_id": "111"}]
+    }
+    # detect_mime_type needs the native SDK — stub it.
+    monkeypatch.setattr(xchat_adapter, "detect_mime_type", lambda b: "image/png")
+
+    await adapter._poll_conversation("111-999")
+
+    assert len(captured) == 1
+    ev = captured[0]
+    assert ev.text == "look at this"
+    assert len(ev.media_urls) == 1
+    assert ev.media_types == ["image/png"]
+    from pathlib import Path as _P
+
+    assert _P(ev.media_urls[0]).read_bytes() == png
+    assert crypto.media_decrypted == [b"ENC" + png]
+
+
+@pytest.mark.asyncio
+async def test_outbound_media_encrypt_upload_attach(monkeypatch, tmp_path):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    adapter._conversation_keys["111:999"] = {"2": b"k2"}
+    adapter._latest_key_version["111:999"] = "2"
+
+    f = tmp_path / "doc.pdf"
+    f.write_bytes(b"%PDF-fake")
+    monkeypatch.setattr(xchat_adapter, "detect_image_dimensions", lambda b: None)
+
+    result = await adapter.send_document("111:999", str(f))
+    assert result.success, result.error
+    # Encrypted before upload...
+    assert crypto.media_encrypted == [b"%PDF-fake"]
+    assert api.uploads and api.uploads[0][1] == b"ENC%PDF-fake"
+    # ...and the send body carried the attachment descriptor.
+    conv, text, attachments = crypto.encrypted[0]
+    assert attachments and attachments[0]["media_hash_key"] == "mhk-1"
+    assert attachments[0]["filename"] == "doc.pdf"
+
+
+@pytest.mark.asyncio
+async def test_outbound_media_without_key_fails_cleanly(monkeypatch, tmp_path):
+    adapter = _make_adapter(monkeypatch)
+    _wire(adapter)
+    f = tmp_path / "a.png"
+    f.write_bytes(b"x")
+    result = await adapter.send_image("111:999", str(f))
+    assert not result.success
+    assert "conversation key" in (result.error or "")
+
+
+@pytest.mark.asyncio
+async def test_native_threaded_reply_uses_cached_event(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    target = {"type": "Message", "id": "e5", "sender_id": "111", "content": {"text": "orig"}}
+    adapter._cache_event("111:999", "e5", target)
+
+    result = await adapter.send("111:999", "threaded answer", reply_to="e5")
+    assert result.success
+    assert crypto.replies == [("111:999", "threaded answer", target, None)]
+    assert crypto.encrypted == []  # took the reply path, not plain encrypt
+
+    # Unknown reply target falls back to a plain send.
+    result = await adapter.send("111:999", "plain", reply_to="nope")
+    assert result.success
+    assert crypto.encrypted == [("111:999", "plain", None)]
+
+
+@pytest.mark.asyncio
+async def test_meta_key_events_absorbed_before_messages(monkeypatch):
+    """KeyChange events arrive in meta.conversation_key_events — they must
+    seed the key cache before this poll's messages are decrypted."""
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._cursors["111-999"] = "e1"
+
+    crypto.decrypt_map["MSG"] = {
+        "type": "Message",
+        "id": "e2",
+        "sender_id": "111",
+        "conversation_id": "111:999",
+        "content": {"text": "post-rotation"},
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e2", "encoded_event": "MSG", "sender_id": "111"}],
+        "meta": {"conversation_key_events": ["KEYCHG"]},
+    }
+    await adapter._poll_conversation("111-999")
+
+    assert crypto.batch_calls == [["KEYCHG"]]
+    assert adapter._conversation_keys["111-999"] == {"1": b"k"}
+    assert adapter._latest_key_version["111-999"] == "1"
+    assert [ev.text for ev in captured] == ["post-rotation"]
+
+
+@pytest.mark.asyncio
+async def test_read_receipts_opt_in(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    _capture(adapter, monkeypatch)
+    adapter._cursors["111-999"] = "e1"
+    crypto.decrypt_map["CCC"] = {
+        "type": "Message",
+        "id": "e3",
+        "sender_id": "111",
+        "conversation_id": "111:999",
+        "content": {"text": "hi"},
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e3", "encoded_event": "CCC", "sender_id": "111", "sequence_id": "sq3"}]
+    }
+
+    # Default: off.
+    await adapter._poll_conversation("111-999")
+    assert api.reads == []
+
+    # Opt in.
+    adapter2 = _make_adapter(monkeypatch, send_read_receipts=True)
+    api2, crypto2 = _wire(adapter2)
+    _capture(adapter2, monkeypatch)
+    adapter2._cursors["111-888"] = "e1"
+    crypto2.decrypt_map["CCC"] = crypto.decrypt_map["CCC"]
+    api2.events_pages["111-888"] = api.events_pages["111-999"]
+    await adapter2._poll_conversation("111-888")
+    assert api2.reads == [("111-888", "sq3")]
+
+
+@pytest.mark.asyncio
+async def test_inbound_reply_context_propagates(monkeypatch):
+    adapter = _make_adapter(monkeypatch)
+    api, crypto = _wire(adapter)
+    captured = _capture(adapter, monkeypatch)
+    adapter._cursors["111-999"] = "e1"
+    crypto.decrypt_map["RPL"] = {
+        "type": "Message",
+        "id": "e7",
+        "sender_id": "111",
+        "conversation_id": "111:999",
+        "content": {"text": "and this one?"},
+        "reply_to": {"id": "e4", "sender_id": "999", "text": "earlier bot answer"},
+    }
+    api.events_pages["111-999"] = {
+        "data": [{"id": "e7", "encoded_event": "RPL", "sender_id": "111"}]
+    }
+    await adapter._poll_conversation("111-999")
+
+    ev = captured[0]
+    assert ev.reply_to_message_id == "e4"
+    assert ev.reply_to_text == "earlier bot answer"
+    assert ev.reply_to_is_own_message is True
+
+
+@pytest.mark.asyncio
+async def test_standalone_new_conversation_handshake(monkeypatch):
+    """Standalone send to a bare user id with no existing conversation
+    performs the verified key handshake, then sends under the fresh key."""
+    monkeypatch.setenv("XCHAT_ACCESS_TOKEN", "tok")
+    monkeypatch.setenv("XCHAT_PRIVATE_KEYS_B64", "YmxvYg==")
+    monkeypatch.setenv("XCHAT_USER_ID", "999")
+
+    api = FakeApi()
+    api.public_keys["999"] = [
+        {"public_key_version": "1", "public_key": "BOT-IPK",
+         "signing_public_key": "BOT-SPK", "identity_public_key_signature": "BOT-SIG"}
+    ]
+    api.public_keys["111"] = [
+        {"public_key_version": "2", "public_key": "USR-IPK",
+         "signing_public_key": "USR-SPK", "identity_public_key_signature": "USR-SIG"}
+    ]
+
+    class HandshakeCrypto(FakeCrypto):
+        def __init__(self):
+            super().__init__()
+            self.bindings: List[tuple] = []
+            self.prepared = False
+            self.explicit_key_used = None
+
+        def load_keys(self, blob, version="1"):
+            pass
+
+        def set_identity(self, user_id):
+            pass
+
+        def set_cache_keys(self, enabled=True):
+            pass
+
+        def verify_key_binding(self, ipk, spk, sig):
+            self.bindings.append((ipk, spk, sig))
+            return True
+
+        def prepare_conversation_key_change(self, public_keys, *, conversation_id=None):
+            self.prepared = True
+            return {
+                "body": {"conversation_key_version": "1",
+                         "conversation_participant_keys": [], "action_signatures": []},
+                "conversation_key": b"fresh-key",
+                "conversation_key_version": "1",
+            }
+
+        def encrypt_text(self, conversation_id, text, *, attachments=None,
+                         conversation_key=None, conversation_key_version=None):
+            if conversation_key is None:
+                raise ValueError("no verified conversation key")
+            self.explicit_key_used = conversation_key
+            return {
+                "message_id": "mid-new",
+                "encoded_message_create_event": "bmV3",
+                "encoded_message_event_signature": "c2ln",
+            }
+
+    crypto = HandshakeCrypto()
+    monkeypatch.setattr(xchat_adapter, "XChatApi", lambda *a, **kw: api)
+    monkeypatch.setattr(xchat_adapter, "XChatCrypto", lambda: crypto)
+
+    cfg = PlatformConfig(enabled=True, extra={})
+    out = await xchat_adapter._standalone_send(cfg, "111", "hello new friend")
+
+    assert out.get("success") is True, out
+    # Both parties' bindings verified, key change POSTed, canonical id adopted.
+    assert len(crypto.bindings) == 2
+    assert crypto.prepared
+    assert api.key_changes and api.key_changes[0][0] == "111"
+    assert crypto.explicit_key_used == b"fresh-key"
+    assert out["chat_id"] == "111:999"
diff --git a/tests/plugins/platforms/xchat/test_xchat_api.py b/tests/plugins/platforms/xchat/test_xchat_api.py
index 4944e43b301a4..af875c0b8d574 100644
--- a/tests/plugins/platforms/xchat/test_xchat_api.py
+++ b/tests/plugins/platforms/xchat/test_xchat_api.py
@@ -183,3 +183,69 @@ async def test_get_events_requests_documented_fields_and_hyphenates_id():
     requested = set((seen["fields"] or "").split(","))
     assert requested and requested <= _VALID_EVENT_FIELDS
     await api.aclose()
+
+
+@pytest.mark.asyncio
+async def test_media_upload_three_step_flow():
+    import base64 as _b64
+
+    calls: List[Dict[str, Any]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        import json as _json
+
+        body = _json.loads(request.content) if request.content else {}
+        calls.append({"path": request.url.path, "body": body})
+        if request.url.path == "/2/chat/media/upload/initialize":
+            assert body["conversation_id"] == "111-999"
+            return httpx.Response(
+                200,
+                json={"data": {"session_id": "sess1", "media_hash_key": "mhk9",
+                               "conversation_id": "111-999"}},
+            )
+        return httpx.Response(200, json={"data": {}})
+
+    api = XChatApi("tok", client=_client(handler))
+    blob = b"E" * (3 * 1024)  # 3 chunks at 1KB chunk size
+    out = await api.media_upload("111:999", blob, chunk_size=1024)
+    assert out == "mhk9"
+
+    paths = [c["path"] for c in calls]
+    assert paths[0] == "/2/chat/media/upload/initialize"
+    appends = [c for c in calls if c["path"].endswith("/append")]
+    assert len(appends) == 3
+    assert [a["body"]["segment_index"] for a in appends] == [0, 1, 2]
+    reassembled = b"".join(_b64.b64decode(a["body"]["media"]) for a in appends)
+    assert reassembled == blob
+    assert paths[-1] == "/2/chat/media/upload/sess1/finalize"
+    await api.aclose()
+
+
+@pytest.mark.asyncio
+async def test_media_download_returns_raw_bytes():
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/2/chat/media/111-999/mhk9"
+        return httpx.Response(200, content=b"\x00ciphertext\xff")
+
+    api = XChatApi("tok", client=_client(handler))
+    blob = await api.media_download("111:999", "mhk9")
+    assert blob == b"\x00ciphertext\xff"
+    await api.aclose()
+
+
+@pytest.mark.asyncio
+async def test_mark_read_posts_sequence_id():
+    seen: Dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        import json as _json
+
+        seen["path"] = request.url.path
+        seen["body"] = _json.loads(request.content)
+        return httpx.Response(200, json={"data": {}})
+
+    api = XChatApi("tok", client=_client(handler))
+    await api.mark_read("111:999", "sq42")
+    assert seen["path"] == "/2/chat/conversations/111-999/read"
+    assert seen["body"] == {"seen_until_sequence_id": "sq42"}
+    await api.aclose()
diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md
index 18d37d555d5b9..0effd6a247ae9 100644
--- a/website/docs/reference/environment-variables.md
+++ b/website/docs/reference/environment-variables.md
@@ -687,6 +687,7 @@ Connect Hermes to [X Chat](https://docs.x.com/xchat/introduction) — X's end-to
 | `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_SEND_READ_RECEIPTS` | Send read receipts for processed messages (`true`/`false`, default `false`). |
 | `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. |
diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md
index 2fbbb591ed3e7..be9c598348621 100644
--- a/website/docs/user-guide/messaging/index.md
+++ b/website/docs/user-guide/messaging/index.md
@@ -47,6 +47,7 @@ Bots need both a model provider and tool providers (TTS, web). A [Nous Portal](/
 | IRC | — | — | — | — | — | — | — |
 | Buzz | — | ✅ | — | ✅ | — | — | — |
 | SimpleX | ✅ | ✅ | ✅ | — | — | ✅ | — |
+| X Chat | ✅ | ✅ | ✅ | — | — | ✅ | — |
 
 **Voice** = TTS audio replies and/or voice message transcription. **Images** = send/receive images. **Files** = send/receive file attachments. **Threads** = threaded conversations. **Reactions** = emoji reactions on messages. **Typing** = typing indicator while processing. **Streaming** = progressive message updates via editing.
 
diff --git a/website/docs/user-guide/messaging/xchat.md b/website/docs/user-guide/messaging/xchat.md
index b2724748d528b..18b5c33b15214 100644
--- a/website/docs/user-guide/messaging/xchat.md
+++ b/website/docs/user-guide/messaging/xchat.md
@@ -7,7 +7,7 @@
 ## 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).
+- 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, and `media.write` to send encrypted file/image attachments).
 - Python 3.10+ (the `chatxdk` E2EE binding is lazy-installed at first use).
 
 ## Setup
@@ -45,6 +45,7 @@ hermes xchat status
 | `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_SEND_READ_RECEIPTS` | Optional | `true` sends read receipts for processed messages (default `false`) |
 | `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 |
@@ -53,8 +54,10 @@ hermes xchat status
 ## How it works
 
 - **Inbound** — the adapter polls each conversation's events endpoint, keeping a **persistent per-conversation cursor** (`~/.hermes/xchat/cursors.json`) of the last processed event. On the very 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**; after that every new event (including bursts larger than one page, and messages that arrived while the gateway was down) is processed exactly once. `KeyChange` events (conversation-key rotations) are verified and folded into the key cache automatically; message **edits** are treated as new messages.
-- **Outbound** — replies are encrypted and signed locally (`encrypt_message` with the session identity), then POSTed as ciphertext.
+- **Outbound** — replies are encrypted and signed locally (`encrypt_message` with the session identity), then POSTed as ciphertext. Replies to a specific message use the native threaded-reply event when the target is in the adapter's decrypted-event cache.
+- **Media** — both directions are fully encrypted. Inbound attachments are downloaded, decrypted with the conversation key for the *event's* key version, and cached locally so vision/file tools can read them. Outbound `MEDIA:<path>` files (images, voice notes, videos, documents) are encrypted with the latest conversation key, uploaded through the 3-step chat-media flow, and attached by `media_hash_key` (requires the `media.write` scope).
 - **Senders** — each new sender's public keys are fetched once and pushed into the XDK's signing-key store so their message signatures verify.
+- **New conversations** — standalone sends (`hermes send xchat:<user-id>`, cron delivery) to a bare numeric user id perform the conversation-key handshake automatically: both parties' key bindings are verified, a fresh conversation key is wrapped for each participant, and the key change is POSTed before the first message.
 - **Identity** — user ids are numeric X user ids; conversation ids look like `123-456` (1:1) or `g123…` (group).
 
 ## Authorization
@@ -95,7 +98,6 @@ Standalone delivery opens an ephemeral E2EE session, seeds the conversation key
 
 ## 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.
+- **Group creation.** The bot participates in existing group conversations but does not create new groups or manage membership.
+- **Access tier.** X Chat API availability depends on your X developer plan; media upload additionally requires the `media.write` scope.