Merge origin/main into fix/discord-allow-channel-threads

Made-with: Cursor

nanobot/channels/feishu.py

@@ -13,6 +13,7 @@ from dataclasses import dataclass
 from typing import Any, Literal
 
 from lark_oapi.api.im.v1.model import MentionEvent, P2ImMessageReceiveV1
+from lark_oapi.core.const import FEISHU_DOMAIN, LARK_DOMAIN
 from loguru import logger
 from pydantic import Field
 
@@ -22,8 +23,6 @@ from nanobot.channels.base import BaseChannel
 from nanobot.config.paths import get_media_dir
 from nanobot.config.schema import Base
 
-from lark_oapi.core.const import FEISHU_DOMAIN, LARK_DOMAIN
-
 FEISHU_AVAILABLE = importlib.util.find_spec("lark_oapi") is not None
 
 # Message type display mapping
@@ -308,6 +307,8 @@ class FeishuChannel(BaseChannel):
         self._loop: asyncio.AbstractEventLoop | None = None
         self._stream_bufs: dict[str, _FeishuStreamBuf] = {}
         self._bot_open_id: str | None = None
+        self._background_tasks: set[asyncio.Task] = set()
+        self._reaction_ids: dict[str, str] = {}  # message_id → reaction_id
 
     @staticmethod
     def _register_optional_event(builder: Any, method_name: str, handler: Any) -> Any:
@@ -549,8 +550,11 @@ class FeishuChannel(BaseChannel):
             return None
 
     async def _add_reaction(self, message_id: str, emoji_type: str = "THUMBSUP") -> str | None:
-        """
-        Add a reaction emoji to a message (non-blocking).
+        """Add a reaction emoji to a message.
+
+        Returns the reaction_id on success, None on failure.
+        When called via a tracked background task, the returned reaction_id
+        is stored in ``_reaction_ids`` for later cleanup by ``send_delta``.
 
         Common emoji types: THUMBSUP, OK, EYES, DONE, OnIt, HEART
         """
@@ -594,6 +598,36 @@ class FeishuChannel(BaseChannel):
         loop = asyncio.get_running_loop()
         await loop.run_in_executor(None, self._remove_reaction_sync, message_id, reaction_id)
 
+    def _on_background_task_done(self, task: asyncio.Task) -> None:
+        """Callback: remove from tracking set and log unhandled exceptions."""
+        self._background_tasks.discard(task)
+        if task.cancelled():
+            return
+        try:
+            task.result()
+        except Exception as exc:
+            logger.warning("Background task failed: {}", exc)
+
+    def _on_reaction_added(self, message_id: str, task: asyncio.Task) -> None:
+        """Callback: store reaction_id after background add-reaction completes."""
+        if task.cancelled():
+            return
+        try:
+            reaction_id = task.result()
+            if reaction_id:
+                self._reaction_ids[message_id] = reaction_id
+        except Exception:
+            pass  # already logged by _on_background_task_done
+        # Trim cache to prevent unbounded growth
+        if len(self._reaction_ids) > 500:
+            self._reaction_ids.pop(next(iter(self._reaction_ids)))
+
+    @staticmethod
+    def _stream_key(chat_id: str, metadata: dict[str, Any] | None = None) -> str:
+        """Scope streaming buffers to the inbound message when available."""
+        meta = metadata or {}
+        return meta.get("message_id") or chat_id
+
     # Regex to match markdown tables (header + separator + data rows)
     _TABLE_RE = re.compile(
         r"((?:^[ \t]*\|.+\|[ \t]*\n)(?:^[ \t]*\|[-:\s|]+\|[ \t]*\n)(?:^[ \t]*\|.+\|[ \t]*\n?)+)",
@@ -1101,17 +1135,23 @@ class FeishuChannel(BaseChannel):
             logger.debug("Feishu: error fetching parent message {}: {}", message_id, e)
             return None
 
-    def _reply_message_sync(self, parent_message_id: str, msg_type: str, content: str) -> bool:
-        """Reply to an existing Feishu message using the Reply API (synchronous)."""
+    def _reply_message_sync(self, parent_message_id: str, msg_type: str, content: str, *, reply_in_thread: bool = False) -> bool:
+        """Reply to an existing Feishu message using the Reply API (synchronous).
+
+        Args:
+            reply_in_thread: If True, reply as a thread/topic message
+                in the Feishu client.
+        """
         from lark_oapi.api.im.v1 import ReplyMessageRequest, ReplyMessageRequestBody
 
         try:
+            body_builder = ReplyMessageRequestBody.builder().msg_type(msg_type).content(content)
+            if reply_in_thread:
+                body_builder = body_builder.reply_in_thread(True)
             request = (
                 ReplyMessageRequest.builder()
                 .message_id(parent_message_id)
-                .request_body(
-                    ReplyMessageRequestBody.builder().msg_type(msg_type).content(content).build()
-                )
+                .request_body(body_builder.build())
                 .build()
             )
             response = self._client.im.v1.message.reply(request)
@@ -1166,8 +1206,19 @@ class FeishuChannel(BaseChannel):
             logger.error("Error sending Feishu {} message: {}", msg_type, e)
             return None
 
-    def _create_streaming_card_sync(self, receive_id_type: str, chat_id: str) -> str | None:
-        """Create a CardKit streaming card, send it to chat, return card_id."""
+    def _create_streaming_card_sync(
+        self,
+        receive_id_type: str,
+        chat_id: str,
+        reply_message_id: str | None = None,
+    ) -> str | None:
+        """Create a CardKit streaming card, send it to chat, return card_id.
+
+        When *reply_message_id* is provided the card is delivered via the
+        reply API (with reply_in_thread=True) so it lands inside the
+        originating thread / topic.  Otherwise the plain create-message
+        API is used.
+        """
         from lark_oapi.api.cardkit.v1 import CreateCardRequest, CreateCardRequestBody
 
         card_json = {
@@ -1196,13 +1247,19 @@ class FeishuChannel(BaseChannel):
                 return None
             card_id = getattr(response.data, "card_id", None)
             if card_id:
-                message_id = self._send_message_sync(
-                    receive_id_type,
-                    chat_id,
-                    "interactive",
-                    json.dumps({"type": "card", "data": {"card_id": card_id}}),
+                card_content = json.dumps(
+                    {"type": "card", "data": {"card_id": card_id}}, ensure_ascii=False
                 )
-                if message_id:
+                if reply_message_id:
+                    sent = self._reply_message_sync(
+                        reply_message_id, "interactive", card_content,
+                        reply_in_thread=True,
+                    )
+                else:
+                    sent = self._send_message_sync(
+                        receive_id_type, chat_id, "interactive", card_content,
+                    ) is not None
+                if sent:
                     return card_id
                 logger.warning(
                     "Created streaming card {} but failed to send it to {}", card_id, chat_id
@@ -1292,23 +1349,27 @@ class FeishuChannel(BaseChannel):
             _stream_end: Finalize the streaming card.
             _tool_hint:  Delta is a formatted tool hint (for display only).
             message_id:  Original message id (used with _stream_end for reaction cleanup).
-            reaction_id: Reaction id to remove on stream end.
+            chat_type:   "group" or "p2p" — controls reply-in-thread for streaming cards.
         """
         if not self._client:
             return
         meta = metadata or {}
+        stream_key = self._stream_key(chat_id, meta)
         loop = asyncio.get_running_loop()
         rid_type = "chat_id" if chat_id.startswith("oc_") else "open_id"
 
         # --- stream end: final update or fallback ---
         if meta.get("_stream_end"):
-            if (message_id := meta.get("message_id")) and (reaction_id := meta.get("reaction_id")):
-                await self._remove_reaction(message_id, reaction_id)
+            message_id = meta.get("message_id")
+            if message_id:
+                reaction_id = self._reaction_ids.pop(message_id, None)
+                if reaction_id:
+                    await self._remove_reaction(message_id, reaction_id)
                 # Add completion emoji if configured
-                if self.config.done_emoji and message_id:
+                if self.config.done_emoji:
                     await self._add_reaction(message_id, self.config.done_emoji)
 
-            buf = self._stream_bufs.pop(chat_id, None)
+            buf = self._stream_bufs.pop(stream_key, None)
             if not buf or not buf.text:
                 return
             # Try to finalize via streaming card; if that fails (e.g.
@@ -1343,24 +1404,45 @@ class FeishuChannel(BaseChannel):
                     {"config": {"wide_screen_mode": True}, "elements": chunk},
                     ensure_ascii=False,
                 )
-                await loop.run_in_executor(
-                    None, self._send_message_sync, rid_type, chat_id, "interactive", card
-                )
+                # Fallback: reply via the Reply API for group chats.
+                # Target message_id — the Feishu API keeps the reply in
+                # the same topic automatically.
+                _f_msg = meta.get("message_id")
+                fallback_msg_id = _f_msg if meta.get("chat_type", "group") == "group" else None
+                if fallback_msg_id:
+                    await loop.run_in_executor(
+                        None, lambda: self._reply_message_sync(
+                            fallback_msg_id, "interactive", card,
+                            reply_in_thread=True,
+                        ),
+                    )
+                else:
+                    await loop.run_in_executor(
+                        None, self._send_message_sync, rid_type, chat_id, "interactive", card
+                    )
             return
 
         # --- accumulate delta ---
-        buf = self._stream_bufs.get(chat_id)
+        buf = self._stream_bufs.get(stream_key)
         if buf is None:
             buf = _FeishuStreamBuf()
-            self._stream_bufs[chat_id] = buf
+            self._stream_bufs[stream_key] = buf
         buf.text += delta
         if not buf.text.strip():
             return
 
         now = time.monotonic()
         if buf.card_id is None:
+            # Send the streaming card as a reply for group chats so it
+            # lands inside the originating topic/thread.  Always target
+            # message_id (the actual inbound message) — the Feishu Reply
+            # API keeps the response in the same topic automatically.
+            is_group = meta.get("chat_type", "group") == "group"
+            reply_msg_id = meta.get("message_id") if is_group else None
             card_id = await loop.run_in_executor(
-                None, self._create_streaming_card_sync, rid_type, chat_id
+                None,
+                self._create_streaming_card_sync,
+                rid_type, chat_id, reply_msg_id,
             )
             if card_id:
                 buf.card_id = card_id
@@ -1393,7 +1475,7 @@ class FeishuChannel(BaseChannel):
                 hint = (msg.content or "").strip()
                 if not hint:
                     return
-                buf = self._stream_bufs.get(msg.chat_id)
+                buf = self._stream_bufs.get(self._stream_key(msg.chat_id, msg.metadata))
                 if buf and buf.card_id:
                     # Delegate to send_delta so tool hints get the same
                     # throttling (and card creation) as regular text deltas.
@@ -1404,37 +1486,59 @@ class FeishuChannel(BaseChannel):
                     return
                 # No active streaming card — send as a regular
                 # interactive card with the same 🔧 prefix style.
+                # Use reply API for group chats so the hint stays in topic.
                 card = json.dumps(
                     {"config": {"wide_screen_mode": True}, "elements": [
                         {"tag": "markdown", "content": self._format_tool_hint_delta(hint)},
                     ]},
                     ensure_ascii=False,
                 )
-                await loop.run_in_executor(
-                    None, self._send_message_sync, receive_id_type, msg.chat_id, "interactive", card
-                )
+                _th_msg_id = msg.metadata.get("message_id")
+                _th_chat_type = msg.metadata.get("chat_type", "group")
+                if _th_msg_id and _th_chat_type == "group":
+                    await loop.run_in_executor(
+                        None, lambda: self._reply_message_sync(
+                            _th_msg_id, "interactive", card,
+                            reply_in_thread=True,
+                        ),
+                    )
+                else:
+                    await loop.run_in_executor(
+                        None, self._send_message_sync, receive_id_type, msg.chat_id, "interactive", card
+                    )
                 return
 
             # Determine whether the first message should quote the user's message.
             # Only the very first send (media or text) in this call uses reply; subsequent
             # chunks/media fall back to plain create to avoid redundant quote bubbles.
+            # Always target message_id — the Feishu Reply API keeps replies in the
+            # same topic automatically when the target message is inside a topic.
             reply_message_id: str | None = None
+            _msg_id = msg.metadata.get("message_id")
             if self.config.reply_to_message and not msg.metadata.get("_progress", False):
-                reply_message_id = msg.metadata.get("message_id") or None
+                reply_message_id = _msg_id
             # For topic group messages, always reply to keep context in thread
             elif msg.metadata.get("thread_id"):
-                reply_message_id = (
-                    msg.metadata.get("root_id") or msg.metadata.get("message_id") or None
-                )
+                reply_message_id = _msg_id
 
             first_send = True  # tracks whether the reply has already been used
 
             def _do_send(m_type: str, content: str) -> None:
-                """Send via reply (first message) or create (subsequent)."""
+                """Send via reply (first message) or create (subsequent).
+
+                For group chats the reply API always uses reply_in_thread=True.
+                The Feishu API automatically keeps replies inside existing
+                topics — reply_in_thread only creates a *new* topic when the
+                target message is a plain (non-topic) message.
+                """
                 nonlocal first_send
                 if reply_message_id and first_send:
                     first_send = False
-                    ok = self._reply_message_sync(reply_message_id, m_type, content)
+                    chat_type = msg.metadata.get("chat_type", "group")
+                    ok = self._reply_message_sync(
+                        reply_message_id, m_type, content,
+                        reply_in_thread=chat_type == "group",
+                    )
                     if ok:
                         return
                     # Fall back to regular send if reply fails
@@ -1457,13 +1561,13 @@ class FeishuChannel(BaseChannel):
                 else:
                     key = await loop.run_in_executor(None, self._upload_file_sync, file_path)
                     if key:
-                        # Use msg_type "audio" for audio, "video" for video, "file" for documents.
+                        # Feishu's OpenAPI names video messages "media".
+                        # Use "audio" for audio, "media" for video, "file" for documents.
                         # Feishu requires these specific msg_types for inline playback.
-                        # Note: "media" is only valid as a tag inside "post" messages, not as a standalone msg_type.
                         if ext in self._AUDIO_EXTS:
                             media_type = "audio"
                         elif ext in self._VIDEO_EXTS:
-                            media_type = "video"
+                            media_type = "media"
                         else:
                             media_type = "file"
                         await loop.run_in_executor(
@@ -1543,8 +1647,13 @@ class FeishuChannel(BaseChannel):
                 logger.debug("Feishu: skipping group message (not mentioned)")
                 return
 
-            # Add reaction
-            reaction_id = await self._add_reaction(message_id, self.config.react_emoji)
+            # Add reaction (non-blocking — tracked background task)
+            task = asyncio.create_task(
+                self._add_reaction(message_id, self.config.react_emoji)
+            )
+            self._background_tasks.add(task)
+            task.add_done_callback(self._on_background_task_done)
+            task.add_done_callback(lambda t: self._on_reaction_added(message_id, t))
 
             # Parse content
             content_parts = []
@@ -1624,6 +1733,15 @@ class FeishuChannel(BaseChannel):
             if not content and not media_paths:
                 return
 
+            # Build topic-scoped session key for conversation isolation.
+            # Group chat: each topic gets its own session via root_id (replies
+            # inside a topic) or message_id (top-level messages start a new topic).
+            # Private chat: no override — same behavior as Telegram/Slack.
+            if chat_type == "group":
+                session_key = f"feishu:{chat_id}:{root_id or message_id}"
+            else:
+                session_key = None
+
             # Forward to message bus
             reply_to = chat_id if chat_type == "group" else sender_id
             await self._handle_message(
@@ -1633,13 +1751,13 @@ class FeishuChannel(BaseChannel):
                 media=media_paths,
                 metadata={
                     "message_id": message_id,
-                    "reaction_id": reaction_id,
                     "chat_type": chat_type,
                     "msg_type": msg_type,
                     "parent_id": parent_id,
                     "root_id": root_id,
                     "thread_id": thread_id,
                 },
+                session_key=session_key,
             )
 
         except Exception as e:

nanobot/channels/manager.py

@@ -172,6 +172,7 @@ class ChannelManager:
                 channel=notice.channel,
                 chat_id=notice.chat_id,
                 content=format_restart_completed_message(notice.started_at_raw),
+                metadata=dict(notice.metadata or {}),
             ),
         ))
 

nanobot/channels/msteams.py

@@ -15,12 +15,21 @@ import asyncio
 import html
 import importlib.util
 import json
+import os
 import re
+import tempfile
 import threading
 import time
+from contextlib import contextmanager
 from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
+
+try:  # pragma: no cover - Windows fallback path
+    import fcntl
+except ImportError:  # pragma: no cover
+    fcntl = None
 
 import httpx
 from loguru import logger
@@ -43,6 +52,13 @@ if TYPE_CHECKING:
 if MSTEAMS_AVAILABLE:
     import jwt
 
+MSTEAMS_REF_TTL_DAYS = 30
+MSTEAMS_REF_TTL_S = MSTEAMS_REF_TTL_DAYS * 24 * 60 * 60
+MSTEAMS_WEBCHAT_HOST = "webchat.botframework.com"
+MSTEAMS_REF_META_FILENAME = "msteams_conversations_meta.json"
+MSTEAMS_REF_LOCK_FILENAME = "msteams_conversations.lock"
+MSTEAMS_REF_TOUCH_INTERVAL_S = 300
+
 
 class MSTeamsConfig(Base):
     """Microsoft Teams channel configuration."""
@@ -58,6 +74,10 @@ class MSTeamsConfig(Base):
     reply_in_thread: bool = True
     mention_only_response: str = "Hi — what can I help with?"
     validate_inbound_auth: bool = True
+    ref_ttl_days: int = Field(default=MSTEAMS_REF_TTL_DAYS, ge=1)
+    prune_web_chat_refs: bool = True
+    prune_non_personal_refs: bool = True
+    ref_touch_interval_s: int = Field(default=MSTEAMS_REF_TOUCH_INTERVAL_S, ge=0)
 
 
 @dataclass
@@ -70,6 +90,7 @@ class ConversationRef:
     activity_id: str | None = None
     conversation_type: str | None = None
     tenant_id: str | None = None
+    updated_at: float | None = None
 
 
 class MSTeamsChannel(BaseChannel):
@@ -102,7 +123,13 @@ class MSTeamsChannel(BaseChannel):
         self._botframework_jwks_expires_at: float = 0.0
         self._refs_path = get_workspace_path() / "state" / "msteams_conversations.json"
         self._refs_path.parent.mkdir(parents=True, exist_ok=True)
+        self._refs_meta_path = self._refs_path.parent / MSTEAMS_REF_META_FILENAME
+        self._refs_lock_path = self._refs_path.parent / MSTEAMS_REF_LOCK_FILENAME
+        self._refs_guard = threading.RLock()
         self._conversation_refs: dict[str, ConversationRef] = self._load_refs()
+        with self._refs_guard:
+            if self._prune_conversation_refs():
+                self._save_refs_locked(prune=True)
 
     async def start(self) -> None:
         """Start the Teams webhook listener."""
@@ -220,7 +247,6 @@ class MSTeamsChannel(BaseChannel):
         token = await self._get_access_token()
         base_url = f"{ref.service_url.rstrip('/')}/v3/conversations/{ref.conversation_id}/activities"
         use_thread_reply = self.config.reply_in_thread and bool(ref.activity_id)
-        url = f"{base_url}/{ref.activity_id}" if use_thread_reply else base_url
         headers = {
             "Authorization": f"Bearer {token}",
             "Content-Type": "application/json",
@@ -233,9 +259,10 @@ class MSTeamsChannel(BaseChannel):
             payload["replyToId"] = ref.activity_id
 
         try:
-            resp = await self._http.post(url, headers=headers, json=payload)
+            resp = await self._http.post(base_url, headers=headers, json=payload)
             resp.raise_for_status()
             logger.info("MSTeams message sent to {}", ref.conversation_id)
+            self._touch_conversation_ref(str(msg.chat_id), persist=True)
         except Exception as e:
             logger.error("MSTeams send failed: {}", e)
             raise
@@ -282,15 +309,17 @@ class MSTeamsChannel(BaseChannel):
             )
             return
 
-        self._conversation_refs[conversation_id] = ConversationRef(
-            service_url=service_url,
-            conversation_id=conversation_id,
-            bot_id=str(recipient.get("id") or "") or None,
-            activity_id=activity_id or None,
-            conversation_type=conversation_type or None,
-            tenant_id=str((channel_data.get("tenant") or {}).get("id") or "") or None,
-        )
-        self._save_refs()
+        with self._refs_guard:
+            self._conversation_refs[conversation_id] = ConversationRef(
+                service_url=service_url,
+                conversation_id=conversation_id,
+                bot_id=str(recipient.get("id") or "") or None,
+                activity_id=activity_id or None,
+                conversation_type=conversation_type or None,
+                tenant_id=str((channel_data.get("tenant") or {}).get("id") or "") or None,
+                updated_at=time.time(),
+            )
+            self._save_refs_locked()
 
         await self._handle_message(
             sender_id=sender_id,
@@ -310,10 +339,12 @@ class MSTeamsChannel(BaseChannel):
         """Extract the user-authored text from a Teams activity."""
         text = str(activity.get("text") or "")
         text = self._strip_possible_bot_mention(text)
+        text = self._normalize_html_whitespace(text)
 
         channel_data = activity.get("channelData") or {}
         reply_to_id = str(activity.get("replyToId") or "").strip()
         normalized_preview = html.unescape(text).replace("&rsquo", "’").strip()
+        normalized_preview = normalized_preview.replace("\xa0", " ")
         normalized_preview = normalized_preview.replace("\r\n", "\n").replace("\r", "\n")
         preview_lines = [line.strip() for line in normalized_preview.split("\n")]
         while preview_lines and not preview_lines[0]:
@@ -333,9 +364,15 @@ class MSTeamsChannel(BaseChannel):
         cleaned = re.sub(r"(?:\r?\n){3,}", "\n\n", cleaned)
         return cleaned.strip()
 
+    def _normalize_html_whitespace(self, text: str) -> str:
+        """Normalize common HTML whitespace/entities from Teams into plain text spacing."""
+        normalized = html.unescape(text).replace("&rsquo", "’")
+        normalized = normalized.replace("\xa0", " ")
+        return normalized
+
     def _normalize_teams_reply_quote(self, text: str) -> str:
         """Normalize Teams quoted replies into a compact structured form."""
-        cleaned = html.unescape(text).replace("&rsquo", "’").strip()
+        cleaned = self._normalize_html_whitespace(text).strip()
         if not cleaned:
             return ""
 
@@ -477,38 +514,243 @@ class MSTeamsChannel(BaseChannel):
         self._botframework_jwks_expires_at = now + 3600
         return self._botframework_jwks
 
-    def _load_refs(self) -> dict[str, ConversationRef]:
-        """Load stored conversation references."""
-        if not self._refs_path.exists():
-            return {}
+    @staticmethod
+    def _safe_float(value: Any) -> float | None:
         try:
-            data = json.loads(self._refs_path.read_text(encoding="utf-8"))
-            out: dict[str, ConversationRef] = {}
-            for key, value in data.items():
-                out[key] = ConversationRef(**value)
-            return out
-        except Exception as e:
-            logger.warning("Failed to load MSTeams conversation refs: {}", e)
+            out = float(value)
+            if out > 0:
+                return out
+        except (TypeError, ValueError):
+            return None
+        return None
+
+    def _normalize_ref_record(self, value: Any) -> ConversationRef | None:
+        """Normalize a stored ref record from legacy/current schema."""
+        if not isinstance(value, dict):
+            return None
+        service_url = str(value.get("service_url") or "").strip()
+        conversation_id = str(value.get("conversation_id") or "").strip()
+        if not service_url or not conversation_id:
+            return None
+        return ConversationRef(
+            service_url=service_url,
+            conversation_id=conversation_id,
+            bot_id=str(value.get("bot_id") or "") or None,
+            activity_id=str(value.get("activity_id") or "") or None,
+            conversation_type=str(value.get("conversation_type") or "") or None,
+            tenant_id=str(value.get("tenant_id") or "") or None,
+            updated_at=self._safe_float(value.get("updated_at")),
+        )
+
+    def _load_refs_raw(self) -> tuple[dict[str, Any], dict[str, Any], bool]:
+        """Load raw refs/main+meta JSON payloads."""
+        main_data: dict[str, Any] = {}
+        meta_data: dict[str, Any] = {}
+        meta_exists = self._refs_meta_path.exists()
+
+        if self._refs_path.exists():
+            try:
+                loaded = json.loads(self._refs_path.read_text(encoding="utf-8"))
+                if isinstance(loaded, dict):
+                    main_data = loaded
+            except Exception as e:
+                logger.warning("Failed to load MSTeams conversation refs: {}", e)
+
+        if meta_exists:
+            try:
+                loaded_meta = json.loads(self._refs_meta_path.read_text(encoding="utf-8"))
+                if isinstance(loaded_meta, dict):
+                    meta_data = loaded_meta
+            except Exception as e:
+                logger.warning("Failed to load MSTeams conversation refs metadata: {}", e)
+
+        return main_data, meta_data, meta_exists
+
+    def _load_refs_from_disk(self) -> dict[str, ConversationRef]:
+        """Load refs from disk with compatibility fallback for legacy layouts."""
+        main_data, meta_data, meta_exists = self._load_refs_raw()
+        if not main_data:
             return {}
 
-    def _save_refs(self) -> None:
-        """Persist conversation references."""
+        out: dict[str, ConversationRef] = {}
+        now = time.time()
+        for key, value in main_data.items():
+            ref = self._normalize_ref_record(value)
+            if not ref:
+                continue
+
+            meta_entry = meta_data.get(key) if isinstance(meta_data, dict) else None
+            meta_ts = None
+            if isinstance(meta_entry, dict):
+                meta_ts = self._safe_float(meta_entry.get("updated_at"))
+            elif meta_entry is not None:
+                meta_ts = self._safe_float(meta_entry)
+
+            if meta_ts is not None:
+                ref.updated_at = meta_ts
+            elif not meta_exists:
+                # First run after introducing meta sidecar: keep legacy refs alive
+                # by initializing timestamps to "now" instead of purging immediately.
+                ref.updated_at = now
+            elif ref.updated_at is None:
+                ref.updated_at = now
+
+            out[key] = ref
+        return out
+
+    def _load_refs(self) -> dict[str, ConversationRef]:
+        """Load stored conversation references."""
+        return self._load_refs_from_disk()
+
+    @contextmanager
+    def _refs_file_lock(self):
+        """Cross-process lock while merging and writing refs state."""
+        self._refs_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_fp = self._refs_lock_path.open("a+", encoding="utf-8")
         try:
-            data = {
-                key: {
-                    "service_url": ref.service_url,
-                    "conversation_id": ref.conversation_id,
-                    "bot_id": ref.bot_id,
-                    "activity_id": ref.activity_id,
-                    "conversation_type": ref.conversation_type,
-                    "tenant_id": ref.tenant_id,
+            if fcntl is not None:
+                fcntl.flock(lock_fp.fileno(), fcntl.LOCK_EX)
+            yield
+        finally:
+            try:
+                if fcntl is not None:
+                    fcntl.flock(lock_fp.fileno(), fcntl.LOCK_UN)
+            finally:
+                lock_fp.close()
+
+    def _is_webchat_service_url(self, service_url: str) -> bool:
+        """Return True when service URL points to unsupported Bot Framework Web Chat."""
+        normalized = service_url.strip()
+        if not normalized:
+            return False
+        host = (urlparse(normalized).hostname or "").strip().lower()
+        if host:
+            return host == MSTEAMS_WEBCHAT_HOST or host.endswith(f".{MSTEAMS_WEBCHAT_HOST}")
+        return MSTEAMS_WEBCHAT_HOST in normalized.lower()
+
+    def _prune_conversation_refs(self, *, now: float | None = None) -> bool:
+        """Remove stale and unsupported conversation refs from memory."""
+        if not self._conversation_refs:
+            return False
+
+        now_ts = time.time() if now is None else now
+        ttl_days = int(self.config.ref_ttl_days)
+        stale_before = now_ts - (ttl_days * 24 * 60 * 60)
+        keys_to_drop: list[str] = []
+
+        for key, ref in self._conversation_refs.items():
+            if self.config.prune_web_chat_refs and self._is_webchat_service_url(ref.service_url):
+                keys_to_drop.append(key)
+                continue
+
+            conv_type = str(ref.conversation_type or "").strip().lower()
+            if self.config.prune_non_personal_refs and conv_type and conv_type != "personal":
+                keys_to_drop.append(key)
+                continue
+
+            try:
+                updated_at = float(ref.updated_at) if ref.updated_at is not None else 0.0
+            except (TypeError, ValueError):
+                updated_at = 0.0
+            if updated_at <= 0 or updated_at < stale_before:
+                keys_to_drop.append(key)
+
+        if not keys_to_drop:
+            return False
+
+        for key in keys_to_drop:
+            self._conversation_refs.pop(key, None)
+        logger.info(
+            "MSTeams pruned {} stale/unsupported conversation refs (ttl={} days)",
+            len(keys_to_drop),
+            ttl_days,
+        )
+        return True
+
+    def _merge_refs_from_disk_locked(self) -> None:
+        """Merge disk refs into memory to reduce lost updates across processes."""
+        disk_refs = self._load_refs_from_disk()
+        for key, disk_ref in disk_refs.items():
+            mem_ref = self._conversation_refs.get(key)
+            if mem_ref is None:
+                self._conversation_refs[key] = disk_ref
+                continue
+            disk_ts = self._safe_float(disk_ref.updated_at) or 0.0
+            mem_ts = self._safe_float(mem_ref.updated_at) or 0.0
+            if disk_ts > mem_ts:
+                self._conversation_refs[key] = disk_ref
+
+    def _touch_conversation_ref(self, chat_id: str, *, persist: bool = False) -> None:
+        """Refresh updated_at for an active ref to keep it from expiring while used."""
+        with self._refs_guard:
+            ref = self._conversation_refs.get(str(chat_id))
+            if not ref:
+                return
+            now = time.time()
+            prev = self._safe_float(ref.updated_at) or 0.0
+            min_interval = max(0, int(self.config.ref_touch_interval_s))
+            if min_interval > 0 and prev > 0 and now - prev < min_interval:
+                return
+            ref.updated_at = now
+            if persist:
+                self._save_refs_locked()
+
+    def _write_json_atomically(self, path, data: dict[str, Any]) -> None:
+        """Write refs JSON atomically to reduce corruption risk during crashes."""
+        payload = json.dumps(data, indent=2)
+        tmp_path: str | None = None
+        try:
+            fd, tmp_path = tempfile.mkstemp(
+                dir=str(path.parent),
+                prefix=f"{path.name}.",
+                suffix=".tmp",
+            )
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                f.write(payload)
+                f.flush()
+                os.fsync(f.fileno())
+            os.replace(tmp_path, path)
+        finally:
+            if tmp_path and os.path.exists(tmp_path):
+                try:
+                    os.unlink(tmp_path)
+                except OSError:
+                    pass
+
+    def _save_refs_locked(self, *, prune: bool = True) -> None:
+        """Persist conversation references (caller must hold _refs_guard)."""
+        try:
+            with self._refs_file_lock():
+                self._merge_refs_from_disk_locked()
+                if prune:
+                    self._prune_conversation_refs()
+                refs_data = {
+                    key: {
+                        "service_url": ref.service_url,
+                        "conversation_id": ref.conversation_id,
+                        "bot_id": ref.bot_id,
+                        "activity_id": ref.activity_id,
+                        "conversation_type": ref.conversation_type,
+                        "tenant_id": ref.tenant_id,
+                    }
+                    for key, ref in self._conversation_refs.items()
                 }
-                for key, ref in self._conversation_refs.items()
-            }
-            self._refs_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+                refs_meta = {
+                    key: {
+                        "updated_at": self._safe_float(ref.updated_at),
+                    }
+                    for key, ref in self._conversation_refs.items()
+                }
+                self._write_json_atomically(self._refs_path, refs_data)
+                self._write_json_atomically(self._refs_meta_path, refs_meta)
         except Exception as e:
             logger.warning("Failed to save MSTeams conversation refs: {}", e)
 
+    def _save_refs(self, *, prune: bool = True) -> None:
+        """Persist conversation references."""
+        with self._refs_guard:
+            self._save_refs_locked(prune=prune)
+
     async def _get_access_token(self) -> str:
         """Fetch an access token for Bot Framework / Azure Bot auth."""
 

nanobot/channels/slack.py

@@ -2,8 +2,10 @@
 
 import asyncio
 import re
+from pathlib import Path
 from typing import Any
 
+import httpx
 from loguru import logger
 from pydantic import Field
 from slack_sdk.socket_mode.request import SocketModeRequest
@@ -15,7 +17,9 @@ from slackify_markdown import slackify_markdown
 from nanobot.bus.events import OutboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
+from nanobot.config.paths import get_media_dir
 from nanobot.config.schema import Base
+from nanobot.utils.helpers import safe_filename, split_message
 
 
 class SlackDMConfig(Base):
@@ -38,12 +42,19 @@ class SlackConfig(Base):
     reply_in_thread: bool = True
     react_emoji: str = "eyes"
     done_emoji: str = "white_check_mark"
+    include_thread_context: bool = True
+    thread_context_limit: int = 20
     allow_from: list[str] = Field(default_factory=list)
     group_policy: str = "mention"
     group_allow_from: list[str] = Field(default_factory=list)
     dm: SlackDMConfig = Field(default_factory=SlackDMConfig)
 
 
+SLACK_MAX_MESSAGE_LEN = 39_000  # Slack API allows ~40k; leave margin
+SLACK_DOWNLOAD_TIMEOUT = 30.0
+_HTML_DOWNLOAD_PREFIXES = (b"<!doctype html", b"<html")
+
+
 class SlackChannel(BaseChannel):
     """Slack channel using Socket Mode."""
 
@@ -57,6 +68,8 @@ class SlackChannel(BaseChannel):
     def default_config(cls) -> dict[str, Any]:
         return SlackConfig().model_dump(by_alias=True)
 
+    _THREAD_CONTEXT_CACHE_LIMIT = 10_000
+
     def __init__(self, config: Any, bus: MessageBus):
         if isinstance(config, dict):
             config = SlackConfig.model_validate(config)
@@ -66,6 +79,7 @@ class SlackChannel(BaseChannel):
         self._socket_client: SocketModeClient | None = None
         self._bot_user_id: str | None = None
         self._target_cache: dict[str, str] = {}
+        self._thread_context_attempted: set[str] = set()
 
     async def start(self) -> None:
         """Start the Slack Socket Mode client."""
@@ -119,23 +133,24 @@ class SlackChannel(BaseChannel):
             target_chat_id = await self._resolve_target_chat_id(msg.chat_id)
             slack_meta = msg.metadata.get("slack", {}) if msg.metadata else {}
             thread_ts = slack_meta.get("thread_ts")
-            channel_type = slack_meta.get("channel_type")
             origin_chat_id = str((slack_meta.get("event", {}) or {}).get("channel") or msg.chat_id)
-            # Slack DMs don't use threads; channel/group replies may keep thread_ts.
-            thread_ts_param = (
-                thread_ts
-                if thread_ts and channel_type != "im" and target_chat_id == origin_chat_id
-                else None
-            )
+            # Reply in the same thread the inbound message belongs to (works
+            # for both real channel threads and DM threads). When the agent
+            # is forwarding to a different channel, drop thread_ts because it
+            # only makes sense within the originating conversation.
+            thread_ts_param = thread_ts if thread_ts and target_chat_id == origin_chat_id else None
 
-            # Slack rejects empty text payloads. Keep media-only messages media-only,
-            # but send a single blank message when the bot has no text or files to send.
             if msg.content or not (msg.media or []):
-                await self._web_client.chat_postMessage(
-                    channel=target_chat_id,
-                    text=self._to_mrkdwn(msg.content) if msg.content else " ",
-                    thread_ts=thread_ts_param,
-                )
+                mrkdwn = self._to_mrkdwn(msg.content) if msg.content else " "
+                buttons = getattr(msg, "buttons", None) or []
+                chunks = split_message(mrkdwn, SLACK_MAX_MESSAGE_LEN)
+                for index, chunk in enumerate(chunks):
+                    kwargs: dict[str, Any] = dict(
+                        channel=target_chat_id, text=chunk, thread_ts=thread_ts_param,
+                    )
+                    if buttons and index == len(chunks) - 1:
+                        kwargs["blocks"] = self._build_button_blocks(chunk, buttons)
+                    await self._web_client.chat_postMessage(**kwargs)
 
             for media_path in msg.media or []:
                 try:
@@ -273,6 +288,9 @@ class SlackChannel(BaseChannel):
         req: SocketModeRequest,
     ) -> None:
         """Handle incoming Socket Mode requests."""
+        if req.type == "interactive":
+            await self._on_block_action(client, req)
+            return
         if req.type != "events_api":
             return
 
@@ -292,8 +310,10 @@ class SlackChannel(BaseChannel):
         sender_id = event.get("user")
         chat_id = event.get("channel")
 
-        # Ignore bot/system messages (any subtype = not a normal user message)
-        if event.get("subtype"):
+        subtype = event.get("subtype")
+        # Slack uses subtype=file_share for user messages with attachments.
+        # Ignore other subtypes such as bot_message / message_changed / deleted.
+        if subtype and subtype != "file_share":
             return
         if self._bot_user_id and sender_id == self._bot_user_id:
             return
@@ -308,7 +328,7 @@ class SlackChannel(BaseChannel):
         logger.debug(
             "Slack event: type={} subtype={} user={} channel={} channel_type={} text={}",
             event_type,
-            event.get("subtype"),
+            subtype,
             sender_id,
             chat_id,
             event.get("channel_type"),
@@ -327,9 +347,18 @@ class SlackChannel(BaseChannel):
 
         text = self._strip_bot_mention(text)
 
-        thread_ts = event.get("thread_ts")
-        if self.config.reply_in_thread and not thread_ts:
-            thread_ts = event.get("ts")
+        event_ts = event.get("ts")
+        raw_thread_ts = event.get("thread_ts")
+        thread_ts = raw_thread_ts
+        # In DMs we don't auto-open a thread on top-level messages (it would
+        # bury replies under "1 reply"). But if the user explicitly opened a
+        # thread inside the DM, raw_thread_ts is set and we honor it.
+        if (
+            self.config.reply_in_thread
+            and not thread_ts
+            and channel_type != "im"
+        ):
+            thread_ts = event_ts
         # Add :eyes: reaction to the triggering message (best-effort)
         try:
             if self._web_client and event.get("ts"):
@@ -341,14 +370,43 @@ class SlackChannel(BaseChannel):
         except Exception as e:
             logger.debug("Slack reactions_add failed: {}", e)
 
-        # Thread-scoped session key for channel/group messages
-        session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts and channel_type != "im" else None
+        # Thread-scoped session key whenever the user is in a real thread
+        # (raw_thread_ts is set). DM threads get their own session, separate
+        # from the DM root, so context doesn't bleed across thread boundaries.
+        session_key = (
+            f"slack:{chat_id}:{thread_ts}" if thread_ts and raw_thread_ts else None
+        )
+        media_paths: list[str] = []
+        file_markers: list[str] = []
+        for file_info in event.get("files") or []:
+            if not isinstance(file_info, dict):
+                continue
+            file_path, marker = await self._download_slack_file(file_info)
+            if file_path:
+                media_paths.append(file_path)
+            if marker:
+                file_markers.append(marker)
+
+        is_slash = text.strip().startswith("/")
+        content = text if is_slash else await self._with_thread_context(
+            text,
+            chat_id=chat_id,
+            channel_type=channel_type,
+            thread_ts=thread_ts,
+            raw_thread_ts=raw_thread_ts,
+            current_ts=event_ts,
+        )
+        if file_markers:
+            content = "\n".join(part for part in [content, *file_markers] if part)
+        if not content and not media_paths:
+            return
 
         try:
             await self._handle_message(
                 sender_id=sender_id,
                 chat_id=chat_id,
-                content=text,
+                content=content,
+                media=media_paths,
                 metadata={
                     "slack": {
                         "event": event,
@@ -361,6 +419,163 @@ class SlackChannel(BaseChannel):
         except Exception:
             logger.exception("Error handling Slack message from {}", sender_id)
 
+    async def _download_slack_file(self, file_info: dict[str, Any]) -> tuple[str | None, str]:
+        """Download a Slack private file to the local media directory."""
+        file_id = str(file_info.get("id") or "file")
+        name = str(
+            file_info.get("name")
+            or file_info.get("title")
+            or file_info.get("id")
+            or "slack-file"
+        )
+        marker_type = "image" if str(file_info.get("mimetype") or "").startswith("image/") else "file"
+        marker = f"[{marker_type}: {name}]"
+        url = str(file_info.get("url_private_download") or file_info.get("url_private") or "")
+        if not url:
+            return None, f"[{marker_type}: {name}: missing download url]"
+        if not self.config.bot_token:
+            return None, f"[{marker_type}: {name}: missing bot token]"
+
+        filename = safe_filename(f"{file_id}_{name}")
+        path = Path(get_media_dir("slack")) / filename
+        try:
+            async with httpx.AsyncClient(timeout=SLACK_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
+                response = await client.get(
+                    url,
+                    headers={"Authorization": f"Bearer {self.config.bot_token}"},
+                )
+                response.raise_for_status()
+            if self._looks_like_html_download(response):
+                raise ValueError("Slack returned HTML instead of file content")
+            path.write_bytes(response.content)
+            return str(path), marker
+        except Exception as e:
+            logger.warning("Failed to download Slack file {}: {}", file_id, e)
+            return None, f"[{marker_type}: {name}: download failed]"
+
+    @staticmethod
+    def _looks_like_html_download(response: httpx.Response) -> bool:
+        content_type = response.headers.get("content-type", "").lower()
+        if "text/html" in content_type:
+            return True
+        preview = response.content[:256].lstrip().lower()
+        return preview.startswith(_HTML_DOWNLOAD_PREFIXES)
+
+    async def _on_block_action(self, client: SocketModeClient, req: SocketModeRequest) -> None:
+        """Handle button clicks from ask_user blocks."""
+        await client.send_socket_mode_response(SocketModeResponse(envelope_id=req.envelope_id))
+        payload = req.payload or {}
+        actions = payload.get("actions") or []
+        if not actions:
+            return
+        value = str(actions[0].get("value") or "")
+        user_info = payload.get("user") or {}
+        sender_id = str(user_info.get("id") or "")
+        channel_info = payload.get("channel") or {}
+        chat_id = str(channel_info.get("id") or "")
+        if not sender_id or not chat_id or not value:
+            return
+        message_info = payload.get("message") or {}
+        thread_ts = message_info.get("thread_ts") or message_info.get("ts")
+        channel_type = self._infer_channel_type(chat_id)
+        if not self._is_allowed(sender_id, chat_id, channel_type):
+            return
+        session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts else None
+        try:
+            await self._handle_message(
+                sender_id=sender_id,
+                chat_id=chat_id,
+                content=value,
+                metadata={"slack": {"thread_ts": thread_ts, "channel_type": channel_type}},
+                session_key=session_key,
+            )
+        except Exception:
+            logger.exception("Error handling Slack button click from {}", sender_id)
+
+    async def _with_thread_context(
+        self,
+        text: str,
+        *,
+        chat_id: str,
+        channel_type: str,
+        thread_ts: str | None,
+        raw_thread_ts: str | None,
+        current_ts: str | None,
+    ) -> str:
+        """Include thread history the first time the bot is pulled into a Slack thread."""
+        del channel_type  # DM and channel threads are both fetched via conversations.replies
+        if (
+            not self.config.include_thread_context
+            or not self._web_client
+            or not raw_thread_ts
+            or not thread_ts
+            or current_ts == thread_ts
+        ):
+            return text
+
+        key = f"{chat_id}:{thread_ts}"
+        if key in self._thread_context_attempted:
+            return text
+        if len(self._thread_context_attempted) >= self._THREAD_CONTEXT_CACHE_LIMIT:
+            self._thread_context_attempted.clear()
+        self._thread_context_attempted.add(key)
+
+        try:
+            response = await self._web_client.conversations_replies(
+                channel=chat_id,
+                ts=thread_ts,
+                limit=max(1, self.config.thread_context_limit),
+            )
+        except Exception as e:
+            logger.warning("Slack thread context unavailable for {}: {}", key, e)
+            return text
+
+        lines = self._format_thread_context(
+            response.get("messages", []),
+            current_ts=current_ts,
+        )
+        if not lines:
+            return text
+        return "Slack thread context before this mention:\n" + "\n".join(lines) + f"\n\nCurrent message:\n{text}"
+
+    def _format_thread_context(self, messages: list[dict[str, Any]], *, current_ts: str | None) -> list[str]:
+        lines: list[str] = []
+        for item in messages:
+            if item.get("ts") == current_ts:
+                continue
+            if item.get("subtype"):
+                continue
+            sender = str(item.get("user") or item.get("bot_id") or "unknown")
+            is_bot = self._bot_user_id is not None and sender == self._bot_user_id
+            label = "bot" if is_bot else f"<@{sender}>"
+            text = str(item.get("text") or "").strip()
+            if not text:
+                continue
+            text = self._strip_bot_mention(text)
+            if len(text) > 500:
+                text = text[:500] + "…"
+            lines.append(f"- {label}: {text}")
+        return lines
+
+    @staticmethod
+    def _build_button_blocks(text: str, buttons: list[list[str]]) -> list[dict[str, Any]]:
+        """Build Slack Block Kit blocks with action buttons for ask_user choices."""
+        blocks: list[dict[str, Any]] = [
+            {"type": "section", "text": {"type": "mrkdwn", "text": text[:3000]}},
+        ]
+        elements = []
+        for row in buttons:
+            for label in row:
+                elements.append({
+                    "type": "button",
+                    "text": {"type": "plain_text", "text": label[:75]},
+                    "value": label[:75],
+                    "action_id": f"ask_user_{label[:50]}",
+                })
+        if elements:
+            blocks.append({"type": "actions", "elements": elements[:25]})
+        return blocks
+
     async def _update_react_emoji(self, chat_id: str, ts: str | None) -> None:
         """Remove the in-progress reaction and optionally add a done reaction."""
         if not self._web_client or not ts:
@@ -407,6 +622,19 @@ class SlackChannel(BaseChannel):
             return chat_id in self.config.group_allow_from
         return False
 
+    def is_allowed(self, sender_id: str) -> bool:
+        # Slack needs channel-aware policy checks, so _on_socket_request and
+        # _on_block_action call _is_allowed before handing off to BaseChannel.
+        return True
+
+    @staticmethod
+    def _infer_channel_type(chat_id: str) -> str:
+        if chat_id.startswith("D"):
+            return "im"
+        if chat_id.startswith("G"):
+            return "group"
+        return "channel"
+
     def _strip_bot_mention(self, text: str) -> str:
         if not text or not self._bot_user_id:
             return text
@@ -425,7 +653,7 @@ class SlackChannel(BaseChannel):
         if not text:
             return ""
         text = cls._TABLE_RE.sub(cls._convert_table, text)
-        return cls._fixup_mrkdwn(slackify_markdown(text))
+        return cls._fixup_mrkdwn(slackify_markdown(text)).rstrip("\n")
 
     @classmethod
     def _fixup_mrkdwn(cls, text: str) -> str:

nanobot/channels/telegram.py

@@ -7,13 +7,14 @@ import re
 import time
 import unicodedata
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Literal
 
 from loguru import logger
 from pydantic import Field
-from telegram import BotCommand, ReactionTypeEmoji, ReplyParameters, Update
+from telegram import BotCommand, InlineKeyboardButton, InlineKeyboardMarkup, ReactionTypeEmoji, ReplyParameters, Update
 from telegram.error import BadRequest, NetworkError, TimedOut
-from telegram.ext import Application, ContextTypes, MessageHandler, filters
+from telegram.ext import Application, CallbackQueryHandler, ContextTypes, MessageHandler, filters
 from telegram.request import HTTPXRequest
 
 from nanobot.bus.events import OutboundMessage
@@ -230,6 +231,8 @@ class TelegramConfig(Base):
     connection_pool_size: int = 32
     pool_timeout: float = 5.0
     streaming: bool = True
+    # Enable inline keyboard buttons in Telegram messages.
+    inline_keyboards: bool = False
     stream_edit_interval: float = Field(default=_STREAM_EDIT_INTERVAL_DEFAULT, ge=0.1)
 
 
@@ -355,15 +358,25 @@ class TelegramChannel(BaseChannel):
         )
         self._app.add_handler(MessageHandler(filters.Regex(r"^/help(?:@\w+)?$"), self._on_help))
 
-        # Add message handler for text, photos, voice, documents, and locations
+        # Add message handler for text, photos, video, voice, documents, and locations
         self._app.add_handler(
             MessageHandler(
-                (filters.TEXT | filters.PHOTO | filters.VOICE | filters.AUDIO | filters.Document.ALL | filters.LOCATION)
+                (filters.TEXT | filters.PHOTO | filters.VIDEO | filters.VIDEO_NOTE
+                 | filters.ANIMATION | filters.VOICE | filters.AUDIO
+                 | filters.Document.ALL | filters.LOCATION)
                 & ~filters.COMMAND,
                 self._on_message
             )
         )
 
+        # Conditionally register inline keyboard callback handler
+        if self.config.inline_keyboards:
+            self._app.add_handler(CallbackQueryHandler(self._on_callback_query))
+            allowed_updates = ["message", "callback_query"]
+            logger.debug("Telegram inline keyboards enabled")
+        else:
+            allowed_updates = ["message"]
+
         logger.info("Starting Telegram bot (polling mode)...")
 
         # Initialize and start polling
@@ -384,7 +397,7 @@ class TelegramChannel(BaseChannel):
 
         # Start polling (this runs until stopped)
         await self._app.updater.start_polling(
-            allowed_updates=["message"],
+            allowed_updates=allowed_updates,
             drop_pending_updates=False,  # Process pending messages on startup
             error_callback=self._on_polling_error,
         )
@@ -419,6 +432,8 @@ class TelegramChannel(BaseChannel):
         ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
         if ext in ("jpg", "jpeg", "png", "gif", "webp"):
             return "photo"
+        if ext in ("mp4", "mov", "avi", "mkv", "webm", "3gp"):
+            return "video"
         if ext == "ogg":
             return "voice"
         if ext in ("mp3", "m4a", "wav", "aac"):
@@ -471,10 +486,19 @@ class TelegramChannel(BaseChannel):
                 media_type = self._get_media_type(media_path)
                 sender = {
                     "photo": self._app.bot.send_photo,
+                    "video": self._app.bot.send_video,
                     "voice": self._app.bot.send_voice,
                     "audio": self._app.bot.send_audio,
                 }.get(media_type, self._app.bot.send_document)
-                param = "photo" if media_type == "photo" else media_type if media_type in ("voice", "audio") else "document"
+                param = {
+                    "photo": "photo",
+                    "video": "video",
+                    "voice": "voice",
+                    "audio": "audio",
+                }.get(media_type, "document")
+                extra: dict[str, Any] = {}
+                if media_type == "video":
+                    extra["supports_streaming"] = True
 
                 # Telegram Bot API accepts HTTP(S) URLs directly for media params.
                 if self._is_remote_media_url(media_path):
@@ -487,16 +511,19 @@ class TelegramChannel(BaseChannel):
                         **{param: media_path},
                         reply_parameters=reply_params,
                         **thread_kwargs,
+                        **extra,
                     )
                     continue
 
-                with open(media_path, "rb") as f:
-                    await sender(
-                        chat_id=chat_id,
-                        **{param: f},
-                        reply_parameters=reply_params,
-                        **thread_kwargs,
-                    )
+                media_bytes = Path(media_path).read_bytes()
+                await self._call_with_retry(
+                    sender,
+                    chat_id=chat_id,
+                    **{param: media_bytes},
+                    reply_parameters=reply_params,
+                    **thread_kwargs,
+                    **extra,
+                )
             except Exception as e:
                 filename = media_path.rsplit("/", 1)[-1]
                 logger.error("Failed to send media {}: {}", media_path, e)
@@ -510,16 +537,25 @@ class TelegramChannel(BaseChannel):
         # Send text content
         if msg.content and msg.content != "[empty message]":
             render_as_blockquote = bool(msg.metadata.get("_tool_hint"))
-            for chunk in split_message(msg.content, TELEGRAM_MAX_MESSAGE_LEN):
+            buttons = getattr(msg, "buttons", None) or []
+            reply_markup = self._build_keyboard(buttons) if buttons else None
+            text = msg.content
+            # Fallback: no native keyboard → splice labels into the message so the choices survive.
+            if buttons and reply_markup is None:
+                text = f"{text}\n\n{self._buttons_as_text(buttons)}"
+            chunks = split_message(text, TELEGRAM_MAX_MESSAGE_LEN)
+            for i, chunk in enumerate(chunks):
+                is_last = (i == len(chunks) - 1)
                 await self._send_text(
                     chat_id, chunk, reply_params, thread_kwargs,
                     render_as_blockquote=render_as_blockquote,
+                    reply_markup=reply_markup if is_last else None,
                 )
 
     async def _call_with_retry(self, fn, *args, **kwargs):
         """Call an async Telegram API function with retry on pool/network timeout and RetryAfter."""
         from telegram.error import RetryAfter
-        
+
         for attempt in range(1, _SEND_MAX_RETRIES + 1):
             try:
                 return await fn(*args, **kwargs)
@@ -549,6 +585,7 @@ class TelegramChannel(BaseChannel):
         reply_params=None,
         thread_kwargs: dict | None = None,
         render_as_blockquote: bool = False,
+        reply_markup=None,
     ) -> None:
         """Send a plain text message with HTML fallback."""
         try:
@@ -557,12 +594,10 @@ class TelegramChannel(BaseChannel):
                 self._app.bot.send_message,
                 chat_id=chat_id, text=html, parse_mode="HTML",
                 reply_parameters=reply_params,
+                reply_markup=reply_markup,
                 **(thread_kwargs or {}),
             )
         except BadRequest as e:
-            # Only fall back to plain text on actual HTML parse/format errors.
-            # Network errors (TimedOut, NetworkError) should propagate immediately
-            # to avoid doubling connection demand during pool exhaustion.
             logger.warning("HTML parse failed, falling back to plain text: {}", e)
             try:
                 await self._call_with_retry(
@@ -570,6 +605,7 @@ class TelegramChannel(BaseChannel):
                     chat_id=chat_id,
                     text=text,
                     reply_parameters=reply_params,
+                    reply_markup=reply_markup,
                     **(thread_kwargs or {}),
                 )
             except Exception as e2:
@@ -796,13 +832,13 @@ class TelegramChannel(BaseChannel):
         text = getattr(reply, "text", None) or getattr(reply, "caption", None) or ""
         if len(text) > TELEGRAM_REPLY_CONTEXT_MAX_LEN:
             text = text[:TELEGRAM_REPLY_CONTEXT_MAX_LEN] + "..."
-            
+
         if not text:
             return None
-            
+
         bot_id, _ = await self._ensure_bot_identity()
         reply_user = getattr(reply, "from_user", None)
-        
+
         if bot_id and reply_user and getattr(reply_user, "id", None) == bot_id:
             return f"[Reply to bot: {text}]"
         elif reply_user and getattr(reply_user, "username", None):
@@ -947,7 +983,7 @@ class TelegramChannel(BaseChannel):
         message = update.message
         user = update.effective_user
         self._remember_thread_context(message)
-        
+
         # Strip @bot_username suffix if present
         content = message.text or ""
         if content.startswith("/") and "@" in content:
@@ -955,7 +991,7 @@ class TelegramChannel(BaseChannel):
             cmd_part = cmd_part.split("@")[0]
             content = f"{cmd_part} {rest[0]}" if rest else cmd_part
         content = self._normalize_telegram_command(content)
-            
+
         await self._handle_message(
             sender_id=self._sender_id(user),
             chat_id=str(message.chat_id),
@@ -1165,18 +1201,76 @@ class TelegramChannel(BaseChannel):
         if mime_type:
             ext_map = {
                 "image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif",
+                "image/webp": ".webp",
                 "audio/ogg": ".ogg", "audio/mpeg": ".mp3", "audio/mp4": ".m4a",
+                "video/mp4": ".mp4", "video/quicktime": ".mov", "video/webm": ".webm",
+                "video/x-matroska": ".mkv", "video/3gpp": ".3gp",
             }
             if mime_type in ext_map:
                 return ext_map[mime_type]
 
-        type_map = {"image": ".jpg", "voice": ".ogg", "audio": ".mp3", "file": ""}
+        type_map = {"image": ".jpg", "voice": ".ogg", "audio": ".mp3", "video": ".mp4", "file": ""}
         if ext := type_map.get(media_type, ""):
             return ext
 
         if filename:
-            from pathlib import Path
-
             return "".join(Path(filename).suffixes)
 
         return ""
+
+    def _build_keyboard(self, buttons: list) -> InlineKeyboardMarkup | None:
+        """Build inline keyboard markup if inline_keyboards is enabled."""
+        if not buttons or not self.config.inline_keyboards:
+            return None
+        keyboard = [
+            [InlineKeyboardButton(label, callback_data=self._safe_callback_data(label)) for label in row]
+            for row in buttons
+        ]
+        return InlineKeyboardMarkup(keyboard)
+
+    @staticmethod
+    def _safe_callback_data(label: str) -> str:
+        # Telegram caps callback_data at 64 bytes UTF-8; truncate at a char boundary so the keyboard still sends.
+        encoded = label.encode("utf-8")
+        if len(encoded) <= 64:
+            return label
+        return encoded[:64].decode("utf-8", errors="ignore")
+
+    @staticmethod
+    def _buttons_as_text(buttons: list[list[str]]) -> str:
+        # Buttons are semantic options; when we can't render a keyboard, the user still needs to see them.
+        return "\n".join(" ".join(f"[{label}]" for label in row) for row in buttons if row)
+
+    async def _on_callback_query(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle inline keyboard button clicks (callback queries)."""
+        if not update.callback_query or not update.effective_user:
+            return
+        query = update.callback_query
+        user = update.effective_user
+        chat_id = query.message.chat_id if query.message else None
+        sender_id = self._sender_id(user)
+        if not chat_id:
+            logger.warning("Callback query without chat_id")
+            return
+        button_label = query.data or ""
+        await query.answer()
+        if query.message:
+            try:
+                await query.message.edit_reply_markup(reply_markup=None)
+            except Exception:
+                pass
+        logger.debug("Inline button tap from {}: {}", sender_id, button_label)
+        self._start_typing(str(chat_id))
+        await self._handle_message(
+            sender_id=sender_id,
+            chat_id=str(chat_id),
+            content=button_label,
+            metadata={
+                "callback_query_id": query.id,
+                "button_label": button_label,
+                "user_id": user.id,
+                "username": user.username,
+                "first_name": user.first_name,
+                "is_callback": True,
+            },
+        )

nanobot/channels/websocket.py

@@ -13,6 +13,7 @@ import json
 import mimetypes
 import re
 import secrets
+import shutil
 import ssl
 import time
 import uuid
@@ -33,6 +34,7 @@ from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.paths import get_media_dir
 from nanobot.config.schema import Base
+from nanobot.utils.helpers import safe_filename
 from nanobot.utils.media_decode import (
     FileSizeExceeded,
     save_base64_data_url,
@@ -52,6 +54,14 @@ def _normalize_config_path(path: str) -> str:
     return _strip_trailing_slash(path)
 
 
+def _append_buttons_as_text(text: str, buttons: list[list[str]]) -> str:
+    labels = [label for row in buttons for label in row if label]
+    if not labels:
+        return text
+    fallback = "\n".join(f"{index}. {label}" for index, label in enumerate(labels, 1))
+    return f"{text}\n\n{fallback}" if text else fallback
+
+
 class WebSocketConfig(Base):
     """WebSocket server channel configuration.
 
@@ -218,12 +228,14 @@ def _parse_envelope(raw: str) -> dict[str, Any] | None:
     return data
 
 
-# Per-message image limits. The server-side guard is a touch looser than the
+# Per-message media limits. The server-side guard is a touch looser than the
 # client's ``Worker`` normalization target (6 MB) — tolerate client slop, but
 # still cap total ingress at ``_MAX_IMAGES_PER_MESSAGE * _MAX_IMAGE_BYTES``
 # which fits comfortably inside ``max_message_bytes``.
 _MAX_IMAGES_PER_MESSAGE = 4
 _MAX_IMAGE_BYTES = 8 * 1024 * 1024
+_MAX_VIDEOS_PER_MESSAGE = 1
+_MAX_VIDEO_BYTES = 20 * 1024 * 1024
 
 # Image MIME whitelist — matches the Composer's ``accept`` list. SVG is
 # explicitly excluded to avoid the XSS surface inside embedded scripts.
@@ -234,6 +246,14 @@ _IMAGE_MIME_ALLOWED: frozenset[str] = frozenset({
     "image/gif",
 })
 
+_VIDEO_MIME_ALLOWED: frozenset[str] = frozenset({
+    "video/mp4",
+    "video/webm",
+    "video/quicktime",
+})
+
+_UPLOAD_MIME_ALLOWED: frozenset[str] = _IMAGE_MIME_ALLOWED | _VIDEO_MIME_ALLOWED
+
 _DATA_URL_MIME_RE = re.compile(r"^data:([^;]+);base64,", re.DOTALL)
 
 
@@ -339,6 +359,9 @@ _MEDIA_ALLOWED_MIMES: frozenset[str] = frozenset({
     "image/jpeg",
     "image/webp",
     "image/gif",
+    "video/mp4",
+    "video/webm",
+    "video/quicktime",
 })
 
 
@@ -516,6 +539,12 @@ class WebSocketChannel(BaseChannel):
         if got == "/api/sessions":
             return self._handle_sessions_list(request)
 
+        if got == "/api/settings":
+            return self._handle_settings(request)
+
+        if got == "/api/settings/update":
+            return self._handle_settings_update(request)
+
         m = re.match(r"^/api/sessions/([^/]+)/messages$", got)
         if m:
             return self._handle_session_messages(request, m.group(1))
@@ -624,6 +653,75 @@ class WebSocketChannel(BaseChannel):
         ]
         return _http_json_response({"sessions": cleaned})
 
+    def _settings_payload(self, *, requires_restart: bool = False) -> dict[str, Any]:
+        from nanobot.config.loader import get_config_path, load_config
+        from nanobot.providers.registry import PROVIDERS, find_by_name
+
+        config = load_config()
+        defaults = config.agents.defaults
+        provider_name = config.get_provider_name(defaults.model) or defaults.provider
+        provider = config.get_provider(defaults.model)
+        selected_provider = provider_name
+        if defaults.provider != "auto":
+            spec = find_by_name(defaults.provider)
+            selected_provider = spec.name if spec else provider_name
+        return {
+            "agent": {
+                "model": defaults.model,
+                "provider": selected_provider,
+                "resolved_provider": provider_name,
+                "has_api_key": bool(provider and provider.api_key),
+            },
+            "providers": [
+                {"name": "auto", "label": "Auto"}
+            ] + [
+                {"name": spec.name, "label": spec.label}
+                for spec in PROVIDERS
+            ],
+            "runtime": {
+                "config_path": str(get_config_path().expanduser()),
+            },
+            "requires_restart": requires_restart,
+        }
+
+    def _handle_settings(self, request: WsRequest) -> Response:
+        if not self._check_api_token(request):
+            return _http_error(401, "Unauthorized")
+        return _http_json_response(self._settings_payload())
+
+    def _handle_settings_update(self, request: WsRequest) -> Response:
+        if not self._check_api_token(request):
+            return _http_error(401, "Unauthorized")
+        from nanobot.config.loader import load_config, save_config
+        from nanobot.providers.registry import find_by_name
+
+        query = _parse_query(request.path)
+        config = load_config()
+        defaults = config.agents.defaults
+        changed = False
+
+        model = _query_first(query, "model")
+        if model is not None:
+            model = model.strip()
+            if not model:
+                return _http_error(400, "model is required")
+            if defaults.model != model:
+                defaults.model = model
+                changed = True
+
+        provider = _query_first(query, "provider")
+        if provider is not None:
+            provider = provider.strip() or "auto"
+            if provider != "auto" and find_by_name(provider) is None:
+                return _http_error(400, "unknown provider")
+            if defaults.provider != provider:
+                defaults.provider = provider
+                changed = True
+
+        if changed:
+            save_config(config)
+        return _http_json_response(self._settings_payload(requires_restart=changed))
+
     @staticmethod
     def _is_webui_session_key(key: str) -> bool:
         """Return True when *key* belongs to the webui's websocket-only surface."""
@@ -703,6 +801,33 @@ class WebSocketChannel(BaseChannel):
         ).digest()[:16]
         return f"/api/media/{_b64url_encode(mac)}/{payload}"
 
+    def _sign_or_stage_media_path(self, path: Path) -> dict[str, str] | None:
+        """Return a signed media URL payload for *path*.
+
+        Persisted inbound media already lives under ``get_media_dir`` and can
+        be signed directly. Outbound bot-generated files may live anywhere on
+        disk; copy those into the websocket media bucket first so the browser
+        can fetch them through the existing signed media route without
+        exposing arbitrary filesystem paths.
+        """
+        signed = self._sign_media_path(path)
+        if signed is not None:
+            return {"url": signed, "name": path.name}
+        try:
+            if not path.is_file():
+                return None
+            media_dir = get_media_dir("websocket")
+            safe_name = safe_filename(path.name) or "attachment"
+            staged = media_dir / f"{uuid.uuid4().hex[:12]}-{safe_name}"
+            shutil.copyfile(path, staged)
+        except OSError as exc:
+            logger.warning("websocket: failed to stage outbound media {}: {}", path, exc)
+            return None
+        signed = self._sign_media_path(staged)
+        if signed is None:
+            return None
+        return {"url": signed, "name": path.name}
+
     def _handle_media_fetch(self, sig: str, payload: str) -> Response:
         """Serve a single media file previously signed via
         :meth:`_sign_media_path`. Validates the signature, decodes the
@@ -945,14 +1070,25 @@ class WebSocketChannel(BaseChannel):
         Returns ``(paths, None)`` on success or ``([], reason)`` on the first
         failure — the caller is expected to surface ``reason`` to the client
         and skip publishing so no half-formed message ever reaches the agent.
-        On failure, any images already written to disk earlier in the same
+        On failure, any files already written to disk earlier in the same
         call are unlinked so partial ingress doesn't leak orphan files.
         ``reason`` is a short, stable token suitable for UI localization.
 
         Shape: ``list[{"data_url": str, "name"?: str | None}]``.
         """
-        if len(media) > _MAX_IMAGES_PER_MESSAGE:
+        image_count = 0
+        video_count = 0
+        for item in media:
+            mime = _extract_data_url_mime(item.get("data_url", "")) if isinstance(item, dict) else None
+            if mime in _VIDEO_MIME_ALLOWED:
+                video_count += 1
+            elif mime in _IMAGE_MIME_ALLOWED:
+                image_count += 1
+        if image_count > _MAX_IMAGES_PER_MESSAGE:
             return [], "too_many_images"
+        if video_count > _MAX_VIDEOS_PER_MESSAGE:
+            return [], "too_many_videos"
+
         media_dir = get_media_dir("websocket")
         paths: list[str] = []
 
@@ -975,11 +1111,13 @@ class WebSocketChannel(BaseChannel):
             mime = _extract_data_url_mime(data_url)
             if mime is None:
                 return _abort("decode")
-            if mime not in _IMAGE_MIME_ALLOWED:
+            if mime not in _UPLOAD_MIME_ALLOWED:
                 return _abort("mime")
+            is_video = mime in _VIDEO_MIME_ALLOWED
+            max_bytes = _MAX_VIDEO_BYTES if is_video else _MAX_IMAGE_BYTES
             try:
                 saved = save_base64_data_url(
-                    data_url, media_dir, max_bytes=_MAX_IMAGE_BYTES,
+                    data_url, media_dir, max_bytes=max_bytes,
                 )
             except FileSizeExceeded:
                 return _abort("size")
@@ -1091,13 +1229,26 @@ class WebSocketChannel(BaseChannel):
         if not conns:
             logger.warning("websocket: no active subscribers for chat_id={}", msg.chat_id)
             return
+        text = msg.content
+        if msg.buttons:
+            text = _append_buttons_as_text(text, msg.buttons)
         payload: dict[str, Any] = {
             "event": "message",
             "chat_id": msg.chat_id,
-            "text": msg.content,
+            "text": text,
         }
+        if msg.buttons:
+            payload["buttons"] = msg.buttons
+            payload["button_prompt"] = msg.content
         if msg.media:
             payload["media"] = msg.media
+            urls: list[dict[str, str]] = []
+            for entry in msg.media:
+                signed = self._sign_or_stage_media_path(Path(entry))
+                if signed is not None:
+                    urls.append(signed)
+            if urls:
+                payload["media_urls"] = urls
         if msg.reply_to:
             payload["reply_to"] = msg.reply_to
         # Mark intermediate agent breadcrumbs (tool-call hints, generic