feat: add IM channels for Feishu, Slack, and Telegram (#1010)

* feat: add IM channels system for Feishu, Slack, and Telegram integration

Bridge external messaging platforms to DeerFlow via LangGraph Server with
async message bus, thread management, and per-channel configuration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address review comments on IM channels system

Fix topic_id handling in store remove/list_entries and manager commands,
correct Telegram reply threading, remove unused imports/variables, update
docstrings and docs to match implementation, and prevent config mutation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* update skill creator

* fix im reply text

* fix comments

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

backend/src/agents/memory/prompt.py

@@ -257,9 +257,7 @@ def format_conversation_for_update(messages: list[Any]) -> str:
         # ephemeral file path info into long-term memory.  Skip the turn entirely
         # when nothing remains after stripping (upload-only message).
         if role == "human":
-            content = re.sub(
-                r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", "", str(content)
-            ).strip()
+            content = re.sub(r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", "", str(content)).strip()
             if not content:
                 continue
 

backend/src/agents/memory/updater.py

@@ -168,11 +168,7 @@ def _strip_upload_mentions_from_memory(memory_data: dict[str, Any]) -> dict[str,
     # Also remove any facts that describe upload events
     facts = memory_data.get("facts", [])
     if facts:
-        memory_data["facts"] = [
-            f
-            for f in facts
-            if not _UPLOAD_SENTENCE_RE.search(f.get("content", ""))
-        ]
+        memory_data["facts"] = [f for f in facts if not _UPLOAD_SENTENCE_RE.search(f.get("content", ""))]
 
     return memory_data
 

backend/src/agents/middlewares/memory_middleware.py

@@ -40,9 +40,7 @@ def _filter_messages_for_memory(messages: list[Any]) -> list[Any]:
     Returns:
         Filtered list containing only user inputs and final assistant responses.
     """
-    _UPLOAD_BLOCK_RE = re.compile(
-        r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", re.IGNORECASE
-    )
+    _UPLOAD_BLOCK_RE = re.compile(r"<uploaded_files>[\s\S]*?</uploaded_files>\n*", re.IGNORECASE)
 
     filtered = []
     skip_next_ai = False
@@ -52,9 +50,7 @@ def _filter_messages_for_memory(messages: list[Any]) -> list[Any]:
         if msg_type == "human":
             content = getattr(msg, "content", "")
             if isinstance(content, list):
-                content = " ".join(
-                    p.get("text", "") for p in content if isinstance(p, dict)
-                )
+                content = " ".join(p.get("text", "") for p in content if isinstance(p, dict))
             content_str = str(content)
             if "<uploaded_files>" in content_str:
                 # Strip the ephemeral upload block; keep the user's real question.

backend/src/channels/__init__.py

@@ -0,0 +1,16 @@
+"""IM Channel integration for DeerFlow.
+
+Provides a pluggable channel system that connects external messaging platforms
+(Feishu/Lark, Slack, Telegram) to the DeerFlow agent via the ChannelManager,
+which uses ``langgraph-sdk`` to communicate with the underlying LangGraph Server.
+"""
+
+from src.channels.base import Channel
+from src.channels.message_bus import InboundMessage, MessageBus, OutboundMessage
+
+__all__ = [
+    "Channel",
+    "InboundMessage",
+    "MessageBus",
+    "OutboundMessage",
+]

backend/src/channels/base.py

@@ -0,0 +1,88 @@
+"""Abstract base class for IM channels."""
+
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+from typing import Any
+
+from src.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage
+
+logger = logging.getLogger(__name__)
+
+
+class Channel(ABC):
+    """Base class for all IM channel implementations.
+
+    Each channel connects to an external messaging platform and:
+    1. Receives messages, wraps them as InboundMessage, publishes to the bus.
+    2. Subscribes to outbound messages and sends replies back to the platform.
+
+    Subclasses must implement ``start``, ``stop``, and ``send``.
+    """
+
+    def __init__(self, name: str, bus: MessageBus, config: dict[str, Any]) -> None:
+        self.name = name
+        self.bus = bus
+        self.config = config
+        self._running = False
+
+    @property
+    def is_running(self) -> bool:
+        return self._running
+
+    # -- lifecycle ---------------------------------------------------------
+
+    @abstractmethod
+    async def start(self) -> None:
+        """Start listening for messages from the external platform."""
+
+    @abstractmethod
+    async def stop(self) -> None:
+        """Gracefully stop the channel."""
+
+    # -- outbound ----------------------------------------------------------
+
+    @abstractmethod
+    async def send(self, msg: OutboundMessage) -> None:
+        """Send a message back to the external platform.
+
+        The implementation should use ``msg.chat_id`` and ``msg.thread_ts``
+        to route the reply to the correct conversation/thread.
+        """
+
+    # -- helpers -----------------------------------------------------------
+
+    def _make_inbound(
+        self,
+        chat_id: str,
+        user_id: str,
+        text: str,
+        *,
+        msg_type: InboundMessageType = InboundMessageType.CHAT,
+        thread_ts: str | None = None,
+        files: list[dict[str, Any]] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> InboundMessage:
+        """Convenience factory for creating InboundMessage instances."""
+        return InboundMessage(
+            channel_name=self.name,
+            chat_id=chat_id,
+            user_id=user_id,
+            text=text,
+            msg_type=msg_type,
+            thread_ts=thread_ts,
+            files=files or [],
+            metadata=metadata or {},
+        )
+
+    async def _on_outbound(self, msg: OutboundMessage) -> None:
+        """Outbound callback registered with the bus.
+
+        Only forwards messages targeted at this channel.
+        """
+        if msg.channel_name == self.name:
+            try:
+                await self.send(msg)
+            except Exception:
+                logger.exception("Failed to send outbound message on channel %s", self.name)

backend/src/channels/feishu.py

@@ -0,0 +1,301 @@
+"""Feishu/Lark channel — connects to Feishu via WebSocket (no public IP needed)."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import threading
+from typing import Any
+
+from src.channels.base import Channel
+from src.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage
+
+logger = logging.getLogger(__name__)
+
+
+class FeishuChannel(Channel):
+    """Feishu/Lark IM channel using the ``lark-oapi`` WebSocket client.
+
+    Configuration keys (in ``config.yaml`` under ``channels.feishu``):
+        - ``app_id``: Feishu app ID.
+        - ``app_secret``: Feishu app secret.
+        - ``verification_token``: (optional) Event verification token.
+
+    The channel uses WebSocket long-connection mode so no public IP is required.
+
+    Message flow:
+        1. User sends a message → bot adds "OK" emoji reaction
+        2. Bot replies in thread: "Working on it......"
+        3. Agent processes the message and returns a result
+        4. Bot replies in thread with the result
+        5. Bot adds "DONE" emoji reaction to the original message
+    """
+
+    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
+        super().__init__(name="feishu", bus=bus, config=config)
+        self._thread: threading.Thread | None = None
+        self._main_loop: asyncio.AbstractEventLoop | None = None
+        self._api_client = None
+        self._CreateMessageReactionRequest = None
+        self._CreateMessageReactionRequestBody = None
+        self._Emoji = None
+
+    async def start(self) -> None:
+        if self._running:
+            return
+
+        try:
+            import lark_oapi as lark
+            from lark_oapi.api.im.v1 import (
+                CreateMessageReactionRequest,
+                CreateMessageReactionRequestBody,
+                CreateMessageRequest,
+                CreateMessageRequestBody,
+                Emoji,
+                ReplyMessageRequest,
+                ReplyMessageRequestBody,
+            )
+        except ImportError:
+            logger.error("lark-oapi is not installed. Install it with: uv add lark-oapi")
+            return
+
+        self._lark = lark
+        self._CreateMessageRequest = CreateMessageRequest
+        self._CreateMessageRequestBody = CreateMessageRequestBody
+        self._ReplyMessageRequest = ReplyMessageRequest
+        self._ReplyMessageRequestBody = ReplyMessageRequestBody
+        self._CreateMessageReactionRequest = CreateMessageReactionRequest
+        self._CreateMessageReactionRequestBody = CreateMessageReactionRequestBody
+        self._Emoji = Emoji
+
+        app_id = self.config.get("app_id", "")
+        app_secret = self.config.get("app_secret", "")
+
+        if not app_id or not app_secret:
+            logger.error("Feishu channel requires app_id and app_secret")
+            return
+
+        self._api_client = lark.Client.builder().app_id(app_id).app_secret(app_secret).build()
+        self._main_loop = asyncio.get_event_loop()
+
+        self._running = True
+        self.bus.subscribe_outbound(self._on_outbound)
+
+        # Both ws.Client construction and start() must happen in a dedicated
+        # thread with its own event loop.  lark-oapi caches the running loop
+        # at construction time and later calls loop.run_until_complete(),
+        # which conflicts with an already-running uvloop.
+        self._thread = threading.Thread(
+            target=self._run_ws,
+            args=(app_id, app_secret),
+            daemon=True,
+        )
+        self._thread.start()
+        logger.info("Feishu channel started")
+
+    def _run_ws(self, app_id: str, app_secret: str) -> None:
+        """Construct and run the lark WS client in a thread with a fresh event loop.
+
+        The lark-oapi SDK captures a module-level event loop at import time
+        (``lark_oapi.ws.client.loop``).  When uvicorn uses uvloop, that
+        captured loop is the *main* thread's uvloop — which is already
+        running, so ``loop.run_until_complete()`` inside ``Client.start()``
+        raises ``RuntimeError``.
+
+        We work around this by creating a plain asyncio event loop for this
+        thread and patching the SDK's module-level reference before calling
+        ``start()``.
+        """
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            import lark_oapi as lark
+            import lark_oapi.ws.client as _ws_client_mod
+
+            # Replace the SDK's module-level loop so Client.start() uses
+            # this thread's (non-running) event loop instead of the main
+            # thread's uvloop.
+            _ws_client_mod.loop = loop
+
+            event_handler = lark.EventDispatcherHandler.builder("", "").register_p2_im_message_receive_v1(self._on_message).build()
+            ws_client = lark.ws.Client(
+                app_id=app_id,
+                app_secret=app_secret,
+                event_handler=event_handler,
+                log_level=lark.LogLevel.INFO,
+            )
+            ws_client.start()
+        except Exception:
+            if self._running:
+                logger.exception("Feishu WebSocket error")
+
+    async def stop(self) -> None:
+        self._running = False
+        self.bus.unsubscribe_outbound(self._on_outbound)
+        if self._thread:
+            self._thread.join(timeout=5)
+            self._thread = None
+        logger.info("Feishu channel stopped")
+
+    async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
+        if not self._api_client:
+            logger.warning("[Feishu] send called but no api_client available")
+            return
+
+        logger.info(
+            "[Feishu] sending reply: chat_id=%s, thread_ts=%s, text_len=%d",
+            msg.chat_id,
+            msg.thread_ts,
+            len(msg.text),
+        )
+        content = self._build_card_content(msg.text)
+
+        last_exc: Exception | None = None
+        for attempt in range(_max_retries):
+            try:
+                if msg.thread_ts:
+                    # Reply in thread (话题)
+                    request = self._ReplyMessageRequest.builder().message_id(msg.thread_ts).request_body(self._ReplyMessageRequestBody.builder().msg_type("interactive").content(content).reply_in_thread(True).build()).build()
+                    await asyncio.to_thread(self._api_client.im.v1.message.reply, request)
+                else:
+                    # Send new message
+                    request = self._CreateMessageRequest.builder().receive_id_type("chat_id").request_body(self._CreateMessageRequestBody.builder().receive_id(msg.chat_id).msg_type("interactive").content(content).build()).build()
+                    await asyncio.to_thread(self._api_client.im.v1.message.create, request)
+
+                # Add "DONE" reaction to the original message on final reply
+                if msg.is_final and msg.thread_ts:
+                    await self._add_reaction(msg.thread_ts, "DONE")
+
+                return  # success
+            except Exception as exc:
+                last_exc = exc
+                if attempt < _max_retries - 1:
+                    delay = 2**attempt  # 1s, 2s
+                    logger.warning(
+                        "[Feishu] send failed (attempt %d/%d), retrying in %ds: %s",
+                        attempt + 1,
+                        _max_retries,
+                        delay,
+                        exc,
+                    )
+                    await asyncio.sleep(delay)
+
+        logger.error("[Feishu] send failed after %d attempts: %s", _max_retries, last_exc)
+        raise last_exc  # type: ignore[misc]
+
+    # -- message formatting ------------------------------------------------
+
+    @staticmethod
+    def _build_card_content(text: str) -> str:
+        """Build a Feishu interactive card with markdown content.
+
+        Feishu's interactive card format natively renders markdown, including
+        headers, bold/italic, code blocks, lists, and links.
+        """
+        card = {
+            "config": {"wide_screen_mode": True},
+            "elements": [{"tag": "markdown", "content": text}],
+        }
+        return json.dumps(card)
+
+    # -- reaction helpers --------------------------------------------------
+
+    async def _add_reaction(self, message_id: str, emoji_type: str = "THUMBSUP") -> None:
+        """Add an emoji reaction to a message."""
+        if not self._api_client or not self._CreateMessageReactionRequest:
+            return
+        try:
+            request = self._CreateMessageReactionRequest.builder().message_id(message_id).request_body(self._CreateMessageReactionRequestBody.builder().reaction_type(self._Emoji.builder().emoji_type(emoji_type).build()).build()).build()
+            await asyncio.to_thread(self._api_client.im.v1.message_reaction.create, request)
+            logger.info("[Feishu] reaction '%s' added to message %s", emoji_type, message_id)
+        except Exception:
+            logger.exception("[Feishu] failed to add reaction '%s' to message %s", emoji_type, message_id)
+
+    async def _send_running_reply(self, message_id: str) -> None:
+        """Reply to a message in-thread with a 'Working on it...' hint."""
+        if not self._api_client:
+            return
+        try:
+            content = self._build_card_content("Working on it...")
+            request = self._ReplyMessageRequest.builder().message_id(message_id).request_body(self._ReplyMessageRequestBody.builder().msg_type("interactive").content(content).reply_in_thread(True).build()).build()
+            await asyncio.to_thread(self._api_client.im.v1.message.reply, request)
+            logger.info("[Feishu] 'Working on it......' reply sent for message %s", message_id)
+        except Exception:
+            logger.exception("[Feishu] failed to send running reply for message %s", message_id)
+
+    # -- internal ----------------------------------------------------------
+
+    @staticmethod
+    def _log_future_error(fut, name: str, msg_id: str) -> None:
+        """Callback for run_coroutine_threadsafe futures to surface errors."""
+        try:
+            exc = fut.exception()
+            if exc:
+                logger.error("[Feishu] %s failed for msg_id=%s: %s", name, msg_id, exc)
+        except Exception:
+            pass
+
+    def _on_message(self, event) -> None:
+        """Called by lark-oapi when a message is received (runs in lark thread)."""
+        try:
+            logger.info("[Feishu] raw event received: type=%s", type(event).__name__)
+            message = event.event.message
+            chat_id = message.chat_id
+            msg_id = message.message_id
+            sender_id = event.event.sender.sender_id.open_id
+
+            # root_id is set when the message is a reply within a Feishu thread.
+            # Use it as topic_id so all replies share the same DeerFlow thread.
+            root_id = getattr(message, "root_id", None) or None
+
+            # Parse message content
+            content = json.loads(message.content)
+            text = content.get("text", "").strip()
+            logger.info(
+                "[Feishu] parsed message: chat_id=%s, msg_id=%s, root_id=%s, sender=%s, text=%r",
+                chat_id,
+                msg_id,
+                root_id,
+                sender_id,
+                text[:100] if text else "",
+            )
+
+            if not text:
+                logger.info("[Feishu] empty text, ignoring message")
+                return
+
+            # Check if it's a command
+            if text.startswith("/"):
+                msg_type = InboundMessageType.COMMAND
+            else:
+                msg_type = InboundMessageType.CHAT
+
+            # topic_id: use root_id for replies (same topic), msg_id for new messages (new topic)
+            topic_id = root_id or msg_id
+
+            inbound = self._make_inbound(
+                chat_id=chat_id,
+                user_id=sender_id,
+                text=text,
+                msg_type=msg_type,
+                thread_ts=msg_id,
+                metadata={"message_id": msg_id, "root_id": root_id},
+            )
+            inbound.topic_id = topic_id
+
+            # Schedule on the async event loop
+            if self._main_loop and self._main_loop.is_running():
+                logger.info("[Feishu] publishing inbound message to bus (type=%s, msg_id=%s)", msg_type.value, msg_id)
+                # Schedule all coroutines and attach error logging to futures
+                for name, coro in [
+                    ("add_reaction", self._add_reaction(msg_id, "OK")),
+                    ("send_running_reply", self._send_running_reply(msg_id)),
+                    ("publish_inbound", self.bus.publish_inbound(inbound)),
+                ]:
+                    fut = asyncio.run_coroutine_threadsafe(coro, self._main_loop)
+                    fut.add_done_callback(lambda f, n=name, mid=msg_id: self._log_future_error(f, n, mid))
+            else:
+                logger.warning("[Feishu] main loop not running, cannot publish inbound message")
+        except Exception:
+            logger.exception("[Feishu] error processing message")

backend/src/channels/manager.py

@@ -0,0 +1,367 @@
+"""ChannelManager — consumes inbound messages and dispatches them to the DeerFlow agent via LangGraph Server."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from src.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage
+from src.channels.store import ChannelStore
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_LANGGRAPH_URL = "http://localhost:2024"
+DEFAULT_GATEWAY_URL = "http://localhost:8001"
+DEFAULT_ASSISTANT_ID = "lead_agent"
+
+
+def _extract_response_text(result: dict | list) -> str:
+    """Extract the last AI message text from a LangGraph runs.wait result.
+
+    ``runs.wait`` returns the final state dict which contains a ``messages``
+    list.  Each message is a dict with at least ``type`` and ``content``.
+
+    Handles special cases:
+    - Regular AI text responses
+    - Clarification interrupts (``ask_clarification`` tool messages)
+    - AI messages with tool_calls but no text content
+    """
+    if isinstance(result, list):
+        messages = result
+    elif isinstance(result, dict):
+        messages = result.get("messages", [])
+    else:
+        return ""
+
+    # Walk backwards to find usable response text
+    for msg in reversed(messages):
+        if not isinstance(msg, dict):
+            continue
+
+        msg_type = msg.get("type")
+
+        # Check for tool messages from ask_clarification (interrupt case)
+        if msg_type == "tool" and msg.get("name") == "ask_clarification":
+            content = msg.get("content", "")
+            if isinstance(content, str) and content:
+                return content
+
+        # Regular AI message with text content
+        if msg_type == "ai":
+            content = msg.get("content", "")
+            if isinstance(content, str) and content:
+                return content
+            # content can be a list of content blocks
+            if isinstance(content, list):
+                parts = []
+                for block in content:
+                    if isinstance(block, dict) and block.get("type") == "text":
+                        parts.append(block.get("text", ""))
+                    elif isinstance(block, str):
+                        parts.append(block)
+                text = "".join(parts)
+                if text:
+                    return text
+    return ""
+
+
+def _extract_artifacts(result: dict | list) -> list[str]:
+    """Extract artifact paths from the last AI response cycle only.
+
+    Instead of reading the full accumulated ``artifacts`` state (which contains
+    all artifacts ever produced in the thread), this inspects the messages after
+    the last human message and collects file paths from ``present_files`` tool
+    calls.  This ensures only newly-produced artifacts are returned.
+    """
+    if isinstance(result, list):
+        messages = result
+    elif isinstance(result, dict):
+        messages = result.get("messages", [])
+    else:
+        return []
+
+    artifacts: list[str] = []
+    for msg in reversed(messages):
+        if not isinstance(msg, dict):
+            continue
+        # Stop at the last human message — anything before it is a previous turn
+        if msg.get("type") == "human":
+            break
+        # Look for AI messages with present_files tool calls
+        if msg.get("type") == "ai":
+            for tc in msg.get("tool_calls", []):
+                if isinstance(tc, dict) and tc.get("name") == "present_files":
+                    args = tc.get("args", {})
+                    paths = args.get("filepaths", [])
+                    if isinstance(paths, list):
+                        artifacts.extend(p for p in paths if isinstance(p, str))
+    return artifacts
+
+
+def _format_artifact_text(artifacts: list[str]) -> str:
+    """Format artifact paths into a human-readable text block listing filenames."""
+    import posixpath
+
+    filenames = [posixpath.basename(p) for p in artifacts]
+    if len(filenames) == 1:
+        return f"Created File: 📎 {filenames[0]}"
+    return "Created Files: 📎 " + "、".join(filenames)
+
+
+class ChannelManager:
+    """Core dispatcher that bridges IM channels to the DeerFlow agent.
+
+    It reads from the MessageBus inbound queue, creates/reuses threads on
+    the LangGraph Server, sends messages via ``runs.wait``, and publishes
+    outbound responses back through the bus.
+    """
+
+    def __init__(
+        self,
+        bus: MessageBus,
+        store: ChannelStore,
+        *,
+        max_concurrency: int = 5,
+        langgraph_url: str = DEFAULT_LANGGRAPH_URL,
+        gateway_url: str = DEFAULT_GATEWAY_URL,
+        assistant_id: str = DEFAULT_ASSISTANT_ID,
+    ) -> None:
+        self.bus = bus
+        self.store = store
+        self._max_concurrency = max_concurrency
+        self._langgraph_url = langgraph_url
+        self._gateway_url = gateway_url
+        self._assistant_id = assistant_id
+        self._client = None  # lazy init — langgraph_sdk async client
+        self._semaphore: asyncio.Semaphore | None = None
+        self._running = False
+        self._task: asyncio.Task | None = None
+
+    # -- LangGraph SDK client (lazy) ----------------------------------------
+
+    def _get_client(self):
+        """Return the ``langgraph_sdk`` async client, creating it on first use."""
+        if self._client is None:
+            from langgraph_sdk import get_client
+
+            self._client = get_client(url=self._langgraph_url)
+        return self._client
+
+    # -- lifecycle ---------------------------------------------------------
+
+    async def start(self) -> None:
+        """Start the dispatch loop."""
+        if self._running:
+            return
+        self._running = True
+        self._semaphore = asyncio.Semaphore(self._max_concurrency)
+        self._task = asyncio.create_task(self._dispatch_loop())
+        logger.info("ChannelManager started (max_concurrency=%d)", self._max_concurrency)
+
+    async def stop(self) -> None:
+        """Stop the dispatch loop."""
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        logger.info("ChannelManager stopped")
+
+    # -- dispatch loop -----------------------------------------------------
+
+    async def _dispatch_loop(self) -> None:
+        logger.info("[Manager] dispatch loop started, waiting for inbound messages")
+        while self._running:
+            try:
+                msg = await asyncio.wait_for(self.bus.get_inbound(), timeout=1.0)
+            except TimeoutError:
+                continue
+            except asyncio.CancelledError:
+                break
+
+            logger.info(
+                "[Manager] received inbound: channel=%s, chat_id=%s, type=%s, text=%r",
+                msg.channel_name,
+                msg.chat_id,
+                msg.msg_type.value,
+                msg.text[:100] if msg.text else "",
+            )
+            task = asyncio.create_task(self._handle_message(msg))
+            task.add_done_callback(self._log_task_error)
+
+    @staticmethod
+    def _log_task_error(task: asyncio.Task) -> None:
+        """Surface unhandled exceptions from background tasks."""
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc:
+            logger.error("[Manager] unhandled error in message task: %s", exc, exc_info=exc)
+
+    async def _handle_message(self, msg: InboundMessage) -> None:
+        async with self._semaphore:
+            try:
+                if msg.msg_type == InboundMessageType.COMMAND:
+                    await self._handle_command(msg)
+                else:
+                    await self._handle_chat(msg)
+            except Exception:
+                logger.exception(
+                    "Error handling message from %s (chat=%s)",
+                    msg.channel_name,
+                    msg.chat_id,
+                )
+                await self._send_error(msg, "An internal error occurred. Please try again.")
+
+    # -- chat handling -----------------------------------------------------
+
+    async def _create_thread(self, client, msg: InboundMessage) -> str:
+        """Create a new thread on the LangGraph Server and store the mapping."""
+        thread = await client.threads.create()
+        thread_id = thread["thread_id"]
+        self.store.set_thread_id(
+            msg.channel_name,
+            msg.chat_id,
+            thread_id,
+            topic_id=msg.topic_id,
+            user_id=msg.user_id,
+        )
+        logger.info("[Manager] new thread created on LangGraph Server: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
+        return thread_id
+
+    async def _handle_chat(self, msg: InboundMessage) -> None:
+        client = self._get_client()
+
+        # Look up existing DeerFlow thread by topic_id (if present)
+        thread_id = None
+        if msg.topic_id:
+            thread_id = self.store.get_thread_id(msg.channel_name, msg.chat_id, topic_id=msg.topic_id)
+            if thread_id:
+                logger.info("[Manager] reusing thread: thread_id=%s for topic_id=%s", thread_id, msg.topic_id)
+
+        # No existing thread found — create a new one
+        if thread_id is None:
+            thread_id = await self._create_thread(client, msg)
+
+        logger.info("[Manager] invoking runs.wait(thread_id=%s, text=%r)", thread_id, msg.text[:100])
+        result = await client.runs.wait(
+            thread_id,
+            self._assistant_id,
+            input={"messages": [{"role": "human", "content": msg.text}]},
+            config={"recursion_limit": 100},
+            context={
+                "thread_id": thread_id,
+                "thinking_enabled": True,
+                "is_plan_mode": False,
+                "subagent_enabled": False,
+            },
+        )
+
+        response_text = _extract_response_text(result)
+        artifacts = _extract_artifacts(result)
+
+        logger.info(
+            "[Manager] agent response received: thread_id=%s, response_len=%d, artifacts=%d",
+            thread_id,
+            len(response_text) if response_text else 0,
+            len(artifacts),
+        )
+
+        # Append artifact filenames when present
+        if artifacts:
+            artifact_text = _format_artifact_text(artifacts)
+            if response_text:
+                response_text = response_text + "\n\n" + artifact_text
+            else:
+                response_text = artifact_text
+
+        if not response_text:
+            response_text = "(No response from agent)"
+
+        outbound = OutboundMessage(
+            channel_name=msg.channel_name,
+            chat_id=msg.chat_id,
+            thread_id=thread_id,
+            text=response_text,
+            artifacts=artifacts,
+            thread_ts=msg.thread_ts,
+        )
+        logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id)
+        await self.bus.publish_outbound(outbound)
+
+    # -- command handling --------------------------------------------------
+
+    async def _handle_command(self, msg: InboundMessage) -> None:
+        text = msg.text.strip()
+        parts = text.split(maxsplit=1)
+        command = parts[0].lower().lstrip("/")
+
+        if command == "new":
+            # Create a new thread on the LangGraph Server
+            client = self._get_client()
+            thread = await client.threads.create()
+            new_thread_id = thread["thread_id"]
+            self.store.set_thread_id(
+                msg.channel_name,
+                msg.chat_id,
+                new_thread_id,
+                topic_id=msg.topic_id,
+                user_id=msg.user_id,
+            )
+            reply = "New conversation started."
+        elif command == "status":
+            thread_id = self.store.get_thread_id(msg.channel_name, msg.chat_id, topic_id=msg.topic_id)
+            reply = f"Active thread: {thread_id}" if thread_id else "No active conversation."
+        elif command == "models":
+            reply = await self._fetch_gateway("/api/models", "models")
+        elif command == "memory":
+            reply = await self._fetch_gateway("/api/memory", "memory")
+        elif command == "help":
+            reply = "Available commands:\n/new — Start a new conversation\n/status — Show current thread info\n/models — List available models\n/memory — Show memory status\n/help — Show this help"
+        else:
+            reply = f"Unknown command: /{command}. Type /help for available commands."
+
+        outbound = OutboundMessage(
+            channel_name=msg.channel_name,
+            chat_id=msg.chat_id,
+            thread_id=self.store.get_thread_id(msg.channel_name, msg.chat_id) or "",
+            text=reply,
+            thread_ts=msg.thread_ts,
+        )
+        await self.bus.publish_outbound(outbound)
+
+    async def _fetch_gateway(self, path: str, kind: str) -> str:
+        """Fetch data from the Gateway API for command responses."""
+        import httpx
+
+        try:
+            async with httpx.AsyncClient() as http:
+                resp = await http.get(f"{self._gateway_url}{path}", timeout=10)
+                resp.raise_for_status()
+                data = resp.json()
+        except Exception:
+            logger.exception("Failed to fetch %s from gateway", kind)
+            return f"Failed to fetch {kind} information."
+
+        if kind == "models":
+            names = [m["name"] for m in data.get("models", [])]
+            return ("Available models:\n" + "\n".join(f"• {n}" for n in names)) if names else "No models configured."
+        elif kind == "memory":
+            facts = data.get("facts", [])
+            return f"Memory contains {len(facts)} fact(s)."
+        return str(data)
+
+    # -- error helper ------------------------------------------------------
+
+    async def _send_error(self, msg: InboundMessage, error_text: str) -> None:
+        outbound = OutboundMessage(
+            channel_name=msg.channel_name,
+            chat_id=msg.chat_id,
+            thread_id=self.store.get_thread_id(msg.channel_name, msg.chat_id) or "",
+            text=error_text,
+            thread_ts=msg.thread_ts,
+        )
+        await self.bus.publish_outbound(outbound)

backend/src/channels/message_bus.py

@@ -0,0 +1,150 @@
+"""MessageBus — async pub/sub hub that decouples channels from the agent dispatcher."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Message types
+# ---------------------------------------------------------------------------
+
+
+class InboundMessageType(StrEnum):
+    """Types of messages arriving from IM channels."""
+
+    CHAT = "chat"
+    COMMAND = "command"
+
+
+@dataclass
+class InboundMessage:
+    """A message arriving from an IM channel toward the agent dispatcher.
+
+    Attributes:
+        channel_name: Name of the source channel (e.g. "feishu", "slack").
+        chat_id: Platform-specific chat/conversation identifier.
+        user_id: Platform-specific user identifier.
+        text: The message text.
+        msg_type: Whether this is a regular chat message or a command.
+        thread_ts: Optional platform thread identifier (for threaded replies).
+        topic_id: Conversation topic identifier used to map to a DeerFlow thread.
+            Messages sharing the same ``topic_id`` within a ``chat_id`` will
+            reuse the same DeerFlow thread.  When ``None``, each message
+            creates a new thread (one-shot Q&A).
+        files: Optional list of file attachments (platform-specific dicts).
+        metadata: Arbitrary extra data from the channel.
+        created_at: Unix timestamp when the message was created.
+    """
+
+    channel_name: str
+    chat_id: str
+    user_id: str
+    text: str
+    msg_type: InboundMessageType = InboundMessageType.CHAT
+    thread_ts: str | None = None
+    topic_id: str | None = None
+    files: list[dict[str, Any]] = field(default_factory=list)
+    metadata: dict[str, Any] = field(default_factory=dict)
+    created_at: float = field(default_factory=time.time)
+
+
+@dataclass
+class OutboundMessage:
+    """A message from the agent dispatcher back to a channel.
+
+    Attributes:
+        channel_name: Target channel name (used for routing).
+        chat_id: Target chat/conversation identifier.
+        thread_id: DeerFlow thread ID that produced this response.
+        text: The response text.
+        artifacts: List of artifact paths produced by the agent.
+        is_final: Whether this is the final message in the response stream.
+        thread_ts: Optional platform thread identifier for threaded replies.
+        metadata: Arbitrary extra data.
+        created_at: Unix timestamp.
+    """
+
+    channel_name: str
+    chat_id: str
+    thread_id: str
+    text: str
+    artifacts: list[str] = field(default_factory=list)
+    is_final: bool = True
+    thread_ts: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+    created_at: float = field(default_factory=time.time)
+
+
+# ---------------------------------------------------------------------------
+# MessageBus
+# ---------------------------------------------------------------------------
+
+OutboundCallback = Callable[[OutboundMessage], Coroutine[Any, Any, None]]
+
+
+class MessageBus:
+    """Async pub/sub hub connecting channels and the agent dispatcher.
+
+    Channels publish inbound messages; the dispatcher consumes them.
+    The dispatcher publishes outbound messages; channels receive them
+    via registered callbacks.
+    """
+
+    def __init__(self) -> None:
+        self._inbound_queue: asyncio.Queue[InboundMessage] = asyncio.Queue()
+        self._outbound_listeners: list[OutboundCallback] = []
+
+    # -- inbound -----------------------------------------------------------
+
+    async def publish_inbound(self, msg: InboundMessage) -> None:
+        """Enqueue an inbound message from a channel."""
+        await self._inbound_queue.put(msg)
+        logger.info(
+            "[Bus] inbound enqueued: channel=%s, chat_id=%s, type=%s, queue_size=%d",
+            msg.channel_name,
+            msg.chat_id,
+            msg.msg_type.value,
+            self._inbound_queue.qsize(),
+        )
+
+    async def get_inbound(self) -> InboundMessage:
+        """Block until the next inbound message is available."""
+        return await self._inbound_queue.get()
+
+    @property
+    def inbound_queue(self) -> asyncio.Queue[InboundMessage]:
+        return self._inbound_queue
+
+    # -- outbound ----------------------------------------------------------
+
+    def subscribe_outbound(self, callback: OutboundCallback) -> None:
+        """Register an async callback for outbound messages."""
+        self._outbound_listeners.append(callback)
+
+    def unsubscribe_outbound(self, callback: OutboundCallback) -> None:
+        """Remove a previously registered outbound callback."""
+        self._outbound_listeners = [cb for cb in self._outbound_listeners if cb is not callback]
+
+    async def publish_outbound(self, msg: OutboundMessage) -> None:
+        """Dispatch an outbound message to all registered listeners."""
+        logger.info(
+            "[Bus] outbound dispatching: channel=%s, chat_id=%s, listeners=%d, text_len=%d",
+            msg.channel_name,
+            msg.chat_id,
+            len(self._outbound_listeners),
+            len(msg.text),
+        )
+        for callback in self._outbound_listeners:
+            try:
+                await callback(msg)
+            except Exception:
+                logger.exception("Error in outbound callback for channel=%s", msg.channel_name)

backend/src/channels/service.py

@@ -0,0 +1,174 @@
+"""ChannelService — manages the lifecycle of all IM channels."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from src.channels.manager import ChannelManager
+from src.channels.message_bus import MessageBus
+from src.channels.store import ChannelStore
+
+logger = logging.getLogger(__name__)
+
+# Channel name → import path for lazy loading
+_CHANNEL_REGISTRY: dict[str, str] = {
+    "feishu": "src.channels.feishu:FeishuChannel",
+    "slack": "src.channels.slack:SlackChannel",
+    "telegram": "src.channels.telegram:TelegramChannel",
+}
+
+
+class ChannelService:
+    """Manages the lifecycle of all configured IM channels.
+
+    Reads configuration from ``config.yaml`` under the ``channels`` key,
+    instantiates enabled channels, and starts the ChannelManager dispatcher.
+    """
+
+    def __init__(self, channels_config: dict[str, Any] | None = None) -> None:
+        self.bus = MessageBus()
+        self.store = ChannelStore()
+        config = dict(channels_config or {})
+        langgraph_url = config.pop("langgraph_url", None) or "http://localhost:2024"
+        gateway_url = config.pop("gateway_url", None) or "http://localhost:8001"
+        self.manager = ChannelManager(
+            bus=self.bus,
+            store=self.store,
+            langgraph_url=langgraph_url,
+            gateway_url=gateway_url,
+        )
+        self._channels: dict[str, Any] = {}  # name -> Channel instance
+        self._config = config
+        self._running = False
+
+    @classmethod
+    def from_app_config(cls) -> ChannelService:
+        """Create a ChannelService from the application config."""
+        from src.config.app_config import get_app_config
+
+        config = get_app_config()
+        channels_config = {}
+        # extra fields are allowed by AppConfig (extra="allow")
+        extra = config.model_extra or {}
+        if "channels" in extra:
+            channels_config = extra["channels"]
+        return cls(channels_config=channels_config)
+
+    async def start(self) -> None:
+        """Start the manager and all enabled channels."""
+        if self._running:
+            return
+
+        await self.manager.start()
+
+        for name, channel_config in self._config.items():
+            if not isinstance(channel_config, dict):
+                continue
+            if not channel_config.get("enabled", False):
+                logger.info("Channel %s is disabled, skipping", name)
+                continue
+
+            await self._start_channel(name, channel_config)
+
+        self._running = True
+        logger.info("ChannelService started with channels: %s", list(self._channels.keys()))
+
+    async def stop(self) -> None:
+        """Stop all channels and the manager."""
+        for name, channel in list(self._channels.items()):
+            try:
+                await channel.stop()
+                logger.info("Channel %s stopped", name)
+            except Exception:
+                logger.exception("Error stopping channel %s", name)
+        self._channels.clear()
+
+        await self.manager.stop()
+        self._running = False
+        logger.info("ChannelService stopped")
+
+    async def restart_channel(self, name: str) -> bool:
+        """Restart a specific channel. Returns True if successful."""
+        if name in self._channels:
+            try:
+                await self._channels[name].stop()
+            except Exception:
+                logger.exception("Error stopping channel %s for restart", name)
+            del self._channels[name]
+
+        config = self._config.get(name)
+        if not config or not isinstance(config, dict):
+            logger.warning("No config for channel %s", name)
+            return False
+
+        return await self._start_channel(name, config)
+
+    async def _start_channel(self, name: str, config: dict[str, Any]) -> bool:
+        """Instantiate and start a single channel."""
+        import_path = _CHANNEL_REGISTRY.get(name)
+        if not import_path:
+            logger.warning("Unknown channel type: %s", name)
+            return False
+
+        try:
+            from src.reflection import resolve_class
+
+            channel_cls = resolve_class(import_path, base_class=None)
+        except Exception:
+            logger.exception("Failed to import channel class for %s", name)
+            return False
+
+        try:
+            channel = channel_cls(bus=self.bus, config=config)
+            await channel.start()
+            self._channels[name] = channel
+            logger.info("Channel %s started", name)
+            return True
+        except Exception:
+            logger.exception("Failed to start channel %s", name)
+            return False
+
+    def get_status(self) -> dict[str, Any]:
+        """Return status information for all channels."""
+        channels_status = {}
+        for name in _CHANNEL_REGISTRY:
+            config = self._config.get(name, {})
+            enabled = isinstance(config, dict) and config.get("enabled", False)
+            running = name in self._channels and self._channels[name].is_running
+            channels_status[name] = {
+                "enabled": enabled,
+                "running": running,
+            }
+        return {
+            "service_running": self._running,
+            "channels": channels_status,
+        }
+
+
+# -- singleton access -------------------------------------------------------
+
+_channel_service: ChannelService | None = None
+
+
+def get_channel_service() -> ChannelService | None:
+    """Get the singleton ChannelService instance (if started)."""
+    return _channel_service
+
+
+async def start_channel_service() -> ChannelService:
+    """Create and start the global ChannelService from app config."""
+    global _channel_service
+    if _channel_service is not None:
+        return _channel_service
+    _channel_service = ChannelService.from_app_config()
+    await _channel_service.start()
+    return _channel_service
+
+
+async def stop_channel_service() -> None:
+    """Stop the global ChannelService."""
+    global _channel_service
+    if _channel_service is not None:
+        await _channel_service.stop()
+        _channel_service = None

backend/src/channels/slack.py

@@ -0,0 +1,223 @@
+"""Slack channel — connects via Socket Mode (no public IP needed)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from markdown_to_mrkdwn import SlackMarkdownConverter
+
+from src.channels.base import Channel
+from src.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage
+
+logger = logging.getLogger(__name__)
+
+_slack_md_converter = SlackMarkdownConverter()
+
+
+class SlackChannel(Channel):
+    """Slack IM channel using Socket Mode (WebSocket, no public IP).
+
+    Configuration keys (in ``config.yaml`` under ``channels.slack``):
+        - ``bot_token``: Slack Bot User OAuth Token (xoxb-...).
+        - ``app_token``: Slack App-Level Token (xapp-...) for Socket Mode.
+        - ``allowed_users``: (optional) List of allowed Slack user IDs. Empty = allow all.
+    """
+
+    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
+        super().__init__(name="slack", bus=bus, config=config)
+        self._socket_client = None
+        self._web_client = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._allowed_users: set[str] = set(config.get("allowed_users", []))
+
+    async def start(self) -> None:
+        if self._running:
+            return
+
+        try:
+            from slack_sdk import WebClient
+            from slack_sdk.socket_mode import SocketModeClient
+            from slack_sdk.socket_mode.response import SocketModeResponse
+        except ImportError:
+            logger.error("slack-sdk is not installed. Install it with: uv add slack-sdk")
+            return
+
+        self._SocketModeResponse = SocketModeResponse
+
+        bot_token = self.config.get("bot_token", "")
+        app_token = self.config.get("app_token", "")
+
+        if not bot_token or not app_token:
+            logger.error("Slack channel requires bot_token and app_token")
+            return
+
+        self._web_client = WebClient(token=bot_token)
+        self._socket_client = SocketModeClient(
+            app_token=app_token,
+            web_client=self._web_client,
+        )
+        self._loop = asyncio.get_event_loop()
+
+        self._socket_client.socket_mode_request_listeners.append(self._on_socket_event)
+
+        self._running = True
+        self.bus.subscribe_outbound(self._on_outbound)
+
+        # Start socket mode in background thread
+        asyncio.get_event_loop().run_in_executor(None, self._socket_client.connect)
+        logger.info("Slack channel started")
+
+    async def stop(self) -> None:
+        self._running = False
+        self.bus.unsubscribe_outbound(self._on_outbound)
+        if self._socket_client:
+            self._socket_client.close()
+            self._socket_client = None
+        logger.info("Slack channel stopped")
+
+    async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
+        if not self._web_client:
+            return
+
+        kwargs: dict[str, Any] = {
+            "channel": msg.chat_id,
+            "text": _slack_md_converter.convert(msg.text),
+        }
+        if msg.thread_ts:
+            kwargs["thread_ts"] = msg.thread_ts
+
+        last_exc: Exception | None = None
+        for attempt in range(_max_retries):
+            try:
+                await asyncio.to_thread(self._web_client.chat_postMessage, **kwargs)
+                # Add a completion reaction to the thread root
+                if msg.thread_ts:
+                    await asyncio.to_thread(
+                        self._add_reaction,
+                        msg.chat_id,
+                        msg.thread_ts,
+                        "white_check_mark",
+                    )
+                return
+            except Exception as exc:
+                last_exc = exc
+                if attempt < _max_retries - 1:
+                    delay = 2**attempt  # 1s, 2s
+                    logger.warning(
+                        "[Slack] send failed (attempt %d/%d), retrying in %ds: %s",
+                        attempt + 1,
+                        _max_retries,
+                        delay,
+                        exc,
+                    )
+                    await asyncio.sleep(delay)
+
+        logger.error("[Slack] send failed after %d attempts: %s", _max_retries, last_exc)
+        # Add failure reaction on error
+        if msg.thread_ts:
+            try:
+                await asyncio.to_thread(
+                    self._add_reaction,
+                    msg.chat_id,
+                    msg.thread_ts,
+                    "x",
+                )
+            except Exception:
+                pass
+        raise last_exc  # type: ignore[misc]
+
+    # -- internal ----------------------------------------------------------
+
+    def _add_reaction(self, channel_id: str, timestamp: str, emoji: str) -> None:
+        """Add an emoji reaction to a message (best-effort, non-blocking)."""
+        if not self._web_client:
+            return
+        try:
+            self._web_client.reactions_add(
+                channel=channel_id,
+                timestamp=timestamp,
+                name=emoji,
+            )
+        except Exception as exc:
+            if "already_reacted" not in str(exc):
+                logger.warning("[Slack] failed to add reaction %s: %s", emoji, exc)
+
+    def _send_running_reply(self, channel_id: str, thread_ts: str) -> None:
+        """Send a 'Working on it......' reply in the thread (called from SDK thread)."""
+        if not self._web_client:
+            return
+        try:
+            self._web_client.chat_postMessage(
+                channel=channel_id,
+                text=":hourglass_flowing_sand: Working on it...",
+                thread_ts=thread_ts,
+            )
+            logger.info("[Slack] 'Working on it...' reply sent in channel=%s, thread_ts=%s", channel_id, thread_ts)
+        except Exception:
+            logger.exception("[Slack] failed to send running reply in channel=%s", channel_id)
+
+    def _on_socket_event(self, client, req) -> None:
+        """Called by slack-sdk for each Socket Mode event."""
+        try:
+            # Acknowledge the event
+            response = self._SocketModeResponse(envelope_id=req.envelope_id)
+            client.send_socket_mode_response(response)
+
+            event_type = req.type
+            if event_type != "events_api":
+                return
+
+            event = req.payload.get("event", {})
+            etype = event.get("type", "")
+
+            # Handle message events (DM or @mention)
+            if etype in ("message", "app_mention"):
+                self._handle_message_event(event)
+
+        except Exception:
+            logger.exception("Error processing Slack event")
+
+    def _handle_message_event(self, event: dict) -> None:
+        # Ignore bot messages
+        if event.get("bot_id") or event.get("subtype"):
+            return
+
+        user_id = event.get("user", "")
+
+        # Check allowed users
+        if self._allowed_users and user_id not in self._allowed_users:
+            logger.debug("Ignoring message from non-allowed user: %s", user_id)
+            return
+
+        text = event.get("text", "").strip()
+        if not text:
+            return
+
+        channel_id = event.get("channel", "")
+        thread_ts = event.get("thread_ts") or event.get("ts", "")
+
+        if text.startswith("/"):
+            msg_type = InboundMessageType.COMMAND
+        else:
+            msg_type = InboundMessageType.CHAT
+
+        # topic_id: use thread_ts as the topic identifier.
+        # For threaded messages, thread_ts is the root message ts (shared topic).
+        # For non-threaded messages, thread_ts is the message's own ts (new topic).
+        inbound = self._make_inbound(
+            chat_id=channel_id,
+            user_id=user_id,
+            text=text,
+            msg_type=msg_type,
+            thread_ts=thread_ts,
+        )
+        inbound.topic_id = thread_ts
+
+        if self._loop and self._loop.is_running():
+            # Acknowledge with an eyes reaction
+            self._add_reaction(channel_id, event.get("ts", thread_ts), "eyes")
+            # Send "running" reply first (fire-and-forget from SDK thread)
+            self._send_running_reply(channel_id, thread_ts)
+            asyncio.run_coroutine_threadsafe(self.bus.publish_inbound(inbound), self._loop)

backend/src/channels/store.py

@@ -0,0 +1,153 @@
+"""ChannelStore — persists IM chat-to-DeerFlow thread mappings."""
+
+from __future__ import annotations
+
+import json
+import logging
+import tempfile
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class ChannelStore:
+    """JSON-file-backed store that maps IM conversations to DeerFlow threads.
+
+    Data layout (on disk)::
+
+        {
+            "<channel_name>:<chat_id>": {
+                "thread_id": "<uuid>",
+                "user_id": "<platform_user>",
+                "created_at": 1700000000.0,
+                "updated_at": 1700000000.0
+            },
+            ...
+        }
+
+    The store is intentionally simple — a single JSON file that is atomically
+    rewritten on every mutation. For production workloads with high concurrency,
+    this can be swapped for a proper database backend.
+    """
+
+    def __init__(self, path: str | Path | None = None) -> None:
+        if path is None:
+            from src.config.paths import get_paths
+
+            path = Path(get_paths().base_dir) / "channels" / "store.json"
+        self._path = Path(path)
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._data: dict[str, dict[str, Any]] = self._load()
+        self._lock = threading.Lock()
+
+    # -- persistence -------------------------------------------------------
+
+    def _load(self) -> dict[str, dict[str, Any]]:
+        if self._path.exists():
+            try:
+                return json.loads(self._path.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError):
+                logger.warning("Corrupt channel store at %s, starting fresh", self._path)
+        return {}
+
+    def _save(self) -> None:
+        fd = tempfile.NamedTemporaryFile(
+            mode="w",
+            dir=self._path.parent,
+            suffix=".tmp",
+            delete=False,
+        )
+        try:
+            json.dump(self._data, fd, indent=2)
+            fd.close()
+            Path(fd.name).replace(self._path)
+        except BaseException:
+            fd.close()
+            Path(fd.name).unlink(missing_ok=True)
+            raise
+
+    # -- key helpers -------------------------------------------------------
+
+    @staticmethod
+    def _key(channel_name: str, chat_id: str, topic_id: str | None = None) -> str:
+        if topic_id:
+            return f"{channel_name}:{chat_id}:{topic_id}"
+        return f"{channel_name}:{chat_id}"
+
+    # -- public API --------------------------------------------------------
+
+    def get_thread_id(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> str | None:
+        """Look up the DeerFlow thread_id for a given IM conversation/topic."""
+        entry = self._data.get(self._key(channel_name, chat_id, topic_id))
+        return entry["thread_id"] if entry else None
+
+    def set_thread_id(
+        self,
+        channel_name: str,
+        chat_id: str,
+        thread_id: str,
+        *,
+        topic_id: str | None = None,
+        user_id: str = "",
+    ) -> None:
+        """Create or update the mapping for an IM conversation/topic."""
+        with self._lock:
+            key = self._key(channel_name, chat_id, topic_id)
+            now = time.time()
+            existing = self._data.get(key)
+            self._data[key] = {
+                "thread_id": thread_id,
+                "user_id": user_id,
+                "created_at": existing["created_at"] if existing else now,
+                "updated_at": now,
+            }
+            self._save()
+
+    def remove(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> bool:
+        """Remove a mapping.
+
+        If ``topic_id`` is provided, only that specific conversation/topic mapping is removed.
+        If ``topic_id`` is omitted, all mappings whose key starts with
+        ``"<channel_name>:<chat_id>"`` (including topic-specific ones) are removed.
+
+        Returns True if at least one mapping was removed.
+        """
+        with self._lock:
+            # Remove a specific conversation/topic mapping.
+            if topic_id is not None:
+                key = self._key(channel_name, chat_id, topic_id)
+                if key in self._data:
+                    del self._data[key]
+                    self._save()
+                    return True
+                return False
+
+            # Remove all mappings for this channel/chat_id (base and any topic-specific keys).
+            prefix = self._key(channel_name, chat_id)
+            keys_to_delete = [k for k in self._data if k == prefix or k.startswith(prefix + ":")]
+            if not keys_to_delete:
+                return False
+
+            for k in keys_to_delete:
+                del self._data[k]
+            self._save()
+            return True
+
+    def list_entries(self, channel_name: str | None = None) -> list[dict[str, Any]]:
+        """List all stored mappings, optionally filtered by channel."""
+        results = []
+        for key, entry in self._data.items():
+            parts = key.split(":", 2)
+            ch = parts[0]
+            chat = parts[1] if len(parts) > 1 else ""
+            topic = parts[2] if len(parts) > 2 else None
+            if channel_name and ch != channel_name:
+                continue
+            item: dict[str, Any] = {"channel_name": ch, "chat_id": chat, **entry}
+            if topic is not None:
+                item["topic_id"] = topic
+            results.append(item)
+        return results

backend/src/channels/telegram.py

@@ -0,0 +1,225 @@
+"""Telegram channel — connects via long-polling (no public IP needed)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+from typing import Any
+
+from src.channels.base import Channel
+from src.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramChannel(Channel):
+    """Telegram bot channel using long-polling.
+
+    Configuration keys (in ``config.yaml`` under ``channels.telegram``):
+        - ``bot_token``: Telegram Bot API token (from @BotFather).
+        - ``allowed_users``: (optional) List of allowed Telegram user IDs. Empty = allow all.
+    """
+
+    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
+        super().__init__(name="telegram", bus=bus, config=config)
+        self._application = None
+        self._thread: threading.Thread | None = None
+        self._tg_loop: asyncio.AbstractEventLoop | None = None
+        self._main_loop: asyncio.AbstractEventLoop | None = None
+        self._allowed_users: set[int] = set()
+        for uid in config.get("allowed_users", []):
+            try:
+                self._allowed_users.add(int(uid))
+            except (ValueError, TypeError):
+                pass
+        # chat_id -> last sent message_id for threaded replies
+        self._last_bot_message: dict[str, int] = {}
+
+    async def start(self) -> None:
+        if self._running:
+            return
+
+        try:
+            from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters
+        except ImportError:
+            logger.error("python-telegram-bot is not installed. Install it with: uv add python-telegram-bot")
+            return
+
+        bot_token = self.config.get("bot_token", "")
+        if not bot_token:
+            logger.error("Telegram channel requires bot_token")
+            return
+
+        self._main_loop = asyncio.get_event_loop()
+        self._running = True
+        self.bus.subscribe_outbound(self._on_outbound)
+
+        # Build the application
+        app = ApplicationBuilder().token(bot_token).build()
+
+        # Command handlers
+        app.add_handler(CommandHandler("start", self._cmd_start))
+        app.add_handler(CommandHandler("new", self._cmd_generic))
+        app.add_handler(CommandHandler("status", self._cmd_generic))
+        app.add_handler(CommandHandler("models", self._cmd_generic))
+        app.add_handler(CommandHandler("memory", self._cmd_generic))
+        app.add_handler(CommandHandler("help", self._cmd_generic))
+
+        # General message handler
+        app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self._on_text))
+
+        self._application = app
+
+        # Run polling in a dedicated thread with its own event loop
+        self._thread = threading.Thread(target=self._run_polling, daemon=True)
+        self._thread.start()
+        logger.info("Telegram channel started")
+
+    async def stop(self) -> None:
+        self._running = False
+        self.bus.unsubscribe_outbound(self._on_outbound)
+        if self._application and self._tg_loop:
+            self._tg_loop.call_soon_threadsafe(self._tg_loop.stop)
+        if self._thread:
+            self._thread.join(timeout=5)
+            self._thread = None
+        self._application = None
+        logger.info("Telegram channel stopped")
+
+    async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
+        if not self._application:
+            return
+
+        try:
+            chat_id = int(msg.chat_id)
+        except (ValueError, TypeError):
+            logger.error("Invalid Telegram chat_id: %s", msg.chat_id)
+            return
+
+        kwargs: dict[str, Any] = {"chat_id": chat_id, "text": msg.text}
+
+        # Reply to the last bot message in this chat for threading
+        reply_to = self._last_bot_message.get(msg.chat_id)
+        if reply_to:
+            kwargs["reply_to_message_id"] = reply_to
+
+        bot = self._application.bot
+        last_exc: Exception | None = None
+        for attempt in range(_max_retries):
+            try:
+                sent = await bot.send_message(**kwargs)
+                self._last_bot_message[msg.chat_id] = sent.message_id
+                return
+            except Exception as exc:
+                last_exc = exc
+                if attempt < _max_retries - 1:
+                    delay = 2**attempt  # 1s, 2s
+                    logger.warning(
+                        "[Telegram] send failed (attempt %d/%d), retrying in %ds: %s",
+                        attempt + 1,
+                        _max_retries,
+                        delay,
+                        exc,
+                    )
+                    await asyncio.sleep(delay)
+
+        logger.error("[Telegram] send failed after %d attempts: %s", _max_retries, last_exc)
+        raise last_exc  # type: ignore[misc]
+
+    # -- helpers -----------------------------------------------------------
+
+    async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None:
+        """Send a 'Working on it...' reply to the user's message."""
+        if not self._application:
+            return
+        try:
+            bot = self._application.bot
+            await bot.send_message(
+                chat_id=int(chat_id),
+                text="Working on it...",
+                reply_to_message_id=reply_to_message_id,
+            )
+            logger.info("[Telegram] 'Working on it...' reply sent in chat=%s", chat_id)
+        except Exception:
+            logger.exception("[Telegram] failed to send running reply in chat=%s", chat_id)
+
+    # -- internal ----------------------------------------------------------
+
+    def _run_polling(self) -> None:
+        """Run telegram polling in a dedicated thread."""
+        self._tg_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._tg_loop)
+        try:
+            self._tg_loop.run_until_complete(self._application.run_polling(close_loop=False))
+        except Exception:
+            if self._running:
+                logger.exception("Telegram polling error")
+
+    def _check_user(self, user_id: int) -> bool:
+        if not self._allowed_users:
+            return True
+        return user_id in self._allowed_users
+
+    async def _cmd_start(self, update, context) -> None:
+        """Handle /start command."""
+        if not self._check_user(update.effective_user.id):
+            return
+        await update.message.reply_text("Welcome to DeerFlow! Send me a message to start a conversation.\nType /help for available commands.")
+
+    async def _cmd_generic(self, update, context) -> None:
+        """Forward slash commands to the channel manager."""
+        if not self._check_user(update.effective_user.id):
+            return
+
+        text = update.message.text
+        chat_id = str(update.effective_chat.id)
+        user_id = str(update.effective_user.id)
+        msg_id = str(update.message.message_id)
+
+        inbound = self._make_inbound(
+            chat_id=chat_id,
+            user_id=user_id,
+            text=text,
+            msg_type=InboundMessageType.COMMAND,
+            thread_ts=msg_id,
+        )
+
+        if self._main_loop and self._main_loop.is_running():
+            asyncio.run_coroutine_threadsafe(self._send_running_reply(chat_id, update.message.message_id), self._main_loop)
+            asyncio.run_coroutine_threadsafe(self.bus.publish_inbound(inbound), self._main_loop)
+
+    async def _on_text(self, update, context) -> None:
+        """Handle regular text messages."""
+        if not self._check_user(update.effective_user.id):
+            return
+
+        text = update.message.text.strip()
+        if not text:
+            return
+
+        chat_id = str(update.effective_chat.id)
+        user_id = str(update.effective_user.id)
+        msg_id = str(update.message.message_id)
+
+        # topic_id: if the user is replying to a bot message, look up
+        # the original topic_id stored for that reply chain.  Otherwise
+        # the current message starts a new topic.
+        reply_to = update.message.reply_to_message
+        if reply_to:
+            topic_id = str(reply_to.message_id)
+        else:
+            topic_id = msg_id
+
+        inbound = self._make_inbound(
+            chat_id=chat_id,
+            user_id=user_id,
+            text=text,
+            msg_type=InboundMessageType.CHAT,
+            thread_ts=msg_id,
+        )
+        inbound.topic_id = topic_id
+
+        if self._main_loop and self._main_loop.is_running():
+            asyncio.run_coroutine_threadsafe(self._send_running_reply(chat_id, update.message.message_id), self._main_loop)
+            asyncio.run_coroutine_threadsafe(self.bus.publish_inbound(inbound), self._main_loop)

backend/src/community/infoquest/infoquest_client.py

@@ -82,8 +82,7 @@ class InfoQuestClient:
                     return response_data["reader_result"]
                 elif "content" in response_data:
                     # Fallback to content field if reader_result is not available
-                    logger.debug("reader_result missing in JSON response, falling back to content field: %s",
-                                 response_data["content"])
+                    logger.debug("reader_result missing in JSON response, falling back to content field: %s", response_data["content"])
                     return response_data["content"]
                 else:
                     # If neither field exists, return the original response

backend/src/gateway/app.py

@@ -10,6 +10,7 @@ from src.gateway.config import get_gateway_config
 from src.gateway.routers import (
     agents,
     artifacts,
+    channels,
     mcp,
     memory,
     models,
@@ -47,7 +48,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # 2. Gateway and LangGraph Server are separate processes with independent caches
     # MCP tools are lazily initialized in LangGraph Server when first needed
 
+    # Start IM channel service if any channels are configured
+    try:
+        from src.channels.service import start_channel_service
+
+        channel_service = await start_channel_service()
+        logger.info("Channel service started: %s", channel_service.get_status())
+    except Exception:
+        logger.exception("No IM channels configured or channel service failed to start")
+
     yield
+
+    # Stop channel service on shutdown
+    try:
+        from src.channels.service import stop_channel_service
+
+        await stop_channel_service()
+    except Exception:
+        logger.exception("Failed to stop channel service")
     logger.info("Shutting down API Gateway")
 
 
@@ -117,6 +135,10 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
                 "name": "suggestions",
                 "description": "Generate follow-up question suggestions for conversations",
             },
+            {
+                "name": "channels",
+                "description": "Manage IM channel integrations (Feishu, Slack, Telegram)",
+            },
             {
                 "name": "health",
                 "description": "Health check and system status endpoints",
@@ -151,6 +173,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)
 
+    # Channels API is mounted at /api/channels
+    app.include_router(channels.router)
+
     @app.get("/health", tags=["health"])
     async def health_check() -> dict:
         """Health check endpoint.

backend/src/gateway/routers/channels.py

@@ -0,0 +1,52 @@
+"""Gateway router for IM channel management."""
+
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/channels", tags=["channels"])
+
+
+class ChannelStatusResponse(BaseModel):
+    service_running: bool
+    channels: dict[str, dict]
+
+
+class ChannelRestartResponse(BaseModel):
+    success: bool
+    message: str
+
+
+@router.get("/", response_model=ChannelStatusResponse)
+async def get_channels_status() -> ChannelStatusResponse:
+    """Get the status of all IM channels."""
+    from src.channels.service import get_channel_service
+
+    service = get_channel_service()
+    if service is None:
+        return ChannelStatusResponse(service_running=False, channels={})
+    status = service.get_status()
+    return ChannelStatusResponse(**status)
+
+
+@router.post("/{name}/restart", response_model=ChannelRestartResponse)
+async def restart_channel(name: str) -> ChannelRestartResponse:
+    """Restart a specific IM channel."""
+    from src.channels.service import get_channel_service
+
+    service = get_channel_service()
+    if service is None:
+        raise HTTPException(status_code=503, detail="Channel service is not running")
+
+    success = await service.restart_channel(name)
+    if success:
+        logger.info("Channel %s restarted successfully", name)
+        return ChannelRestartResponse(success=True, message=f"Channel {name} restarted successfully")
+    else:
+        logger.warning("Failed to restart channel %s", name)
+        return ChannelRestartResponse(success=False, message=f"Failed to restart channel {name}")

backend/src/reflection/resolvers.py

@@ -19,10 +19,7 @@ def _build_missing_dependency_hint(module_path: str, err: ImportError) -> str:
     if package_name is None:
         package_name = MODULE_TO_PACKAGE_HINTS.get(missing_module, missing_module.replace("_", "-"))
 
-    return (
-        f"Missing dependency '{missing_module}'. "
-        f"Install it with `uv add {package_name}` (or `pip install {package_name}`), then restart DeerFlow."
-    )
+    return f"Missing dependency '{missing_module}'. Install it with `uv add {package_name}` (or `pip install {package_name}`), then restart DeerFlow."
 
 
 def resolve_variable[T](

backend/src/sandbox/local/local_sandbox.py

@@ -147,10 +147,7 @@ class LocalSandbox(Sandbox):
         shell_from_path = shutil.which("sh")
         if shell_from_path is not None:
             return shell_from_path
-        raise RuntimeError(
-            "No suitable shell executable found. Tried /bin/zsh, /bin/bash, "
-            "/bin/sh, and `sh` on PATH."
-        )
+        raise RuntimeError("No suitable shell executable found. Tried /bin/zsh, /bin/bash, /bin/sh, and `sh` on PATH.")
 
     def execute_command(self, command: str) -> str:
         # Resolve container paths in command before execution

backend/src/tools/builtins/present_file_tool.py

@@ -54,9 +54,7 @@ def _normalize_presented_filepath(
     try:
         relative_path = actual_path.relative_to(outputs_dir)
     except ValueError as exc:
-        raise ValueError(
-            f"Only files in {OUTPUTS_VIRTUAL_PREFIX} can be presented: {filepath}"
-        ) from exc
+        raise ValueError(f"Only files in {OUTPUTS_VIRTUAL_PREFIX} can be presented: {filepath}") from exc
 
     return f"{OUTPUTS_VIRTUAL_PREFIX}/{relative_path.as_posix()}"
 
@@ -87,22 +85,16 @@ def present_file_tool(
         filepaths: List of absolute file paths to present to the user. **Only** files in `/mnt/user-data/outputs` can be presented.
     """
     try:
-        normalized_paths = [
-            _normalize_presented_filepath(runtime, filepath) for filepath in filepaths
-        ]
+        normalized_paths = [_normalize_presented_filepath(runtime, filepath) for filepath in filepaths]
     except ValueError as exc:
         return Command(
-            update={
-                "messages": [ToolMessage(f"Error: {exc}", tool_call_id=tool_call_id)]
-            },
+            update={"messages": [ToolMessage(f"Error: {exc}", tool_call_id=tool_call_id)]},
         )
 
     # The merge_artifacts reducer will handle merging and deduplication
     return Command(
         update={
             "artifacts": normalized_paths,
-            "messages": [
-                ToolMessage("Successfully presented files", tool_call_id=tool_call_id)
-            ],
+            "messages": [ToolMessage("Successfully presented files", tool_call_id=tool_call_id)],
         },
     )