feat(backend): queue rapid same-thread messages and preserve topic card previews (#3988)

* feat(backend): queue rapid same-thread messages and preserve topic card previews

* fix: checkstyle

* refactor(channels): make feishu serialization policy-driven and fix queue cleanup

README.md

@@ -453,6 +453,7 @@ Notes:
 - `assistant_id: lead_agent` calls the default LangGraph assistant directly.
 - If `assistant_id` is set to a custom agent name, DeerFlow still routes through `lead_agent` and injects that value as `agent_name`, so the custom agent's SOUL/config takes effect for IM channels.
 - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation.
+- Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches.
 
 Set the corresponding API keys in your `.env` file:
 

backend/AGENTS.md

@@ -485,7 +485,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 **Components**:
 - `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels)
 - `store.py` - JSON-file persistence mapping `channel_name:chat_id[:topic_id]` → `thread_id` (keys are `channel:chat` for root conversations and `channel:chat:topic` for threaded conversations)
-- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
+- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
 - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
 - `service.py` - Manages lifecycle of all configured channels from `config.yaml`
 - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` registers the "Working on it..." placeholder as the stream target and edits it in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
@@ -502,7 +502,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound
 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound. The agent posts its own reply mid-run via `gh` from the sandbox. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub).
-7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement)
+7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement). Messages already sent inside an existing Feishu topic carry a compact source-message preview in that card, and queued same-thread follow-ups patch their own source message's card from queued → running → final without falling back to the generic busy reply.
 8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages
 9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails
 10. For commands (`/new`, `/status`, `/models`, `/memory`, `/goal`, `/help`): handle locally or query Gateway API

backend/README.md

@@ -128,9 +128,9 @@ FastAPI application providing REST endpoints for frontend integration:
 
 ### IM Channels
 
-The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])` and updates a single in-thread card in place.
+The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])`, serializes rapid same-thread turns inside the channel manager, and updates a single in-thread card per source message in place.
 
-For Feishu card updates, DeerFlow stores the running card's `message_id` per inbound message and patches that same card until the run finishes, preserving the existing `OK` / `DONE` reaction flow.
+For Feishu card updates, DeerFlow stores the running card's `message_id` per inbound message and patches that same card until the run finishes, preserving the existing `OK` / `DONE` reaction flow. When a follow-up arrives inside an existing Feishu topic while another turn is still running, the later message now waits on the mapped DeerFlow `thread_id`, receives a queued/running card on that exact source message, and keeps a compact source-message blockquote in subsequent patches so rapid consecutive questions remain distinguishable.
 
 ---
 

backend/app/channels/feishu.py

@@ -29,6 +29,7 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 logger = logging.getLogger(__name__)
 PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60
 FEISHU_INBOUND_BATCH_WINDOW_SECONDS = 0.75
+SOURCE_PREVIEW_METADATA_KEY = "feishu_source_preview"
 
 
 def _is_feishu_command(text: str) -> bool:
@@ -85,6 +86,45 @@ class FeishuChannel(Channel):
     def _pending_key(chat_id: str, user_id: str) -> tuple[str, str]:
         return (chat_id, user_id)
 
+    @staticmethod
+    def _should_include_source_preview(
+        *,
+        chat_type: str | None,
+        root_id: str | None,
+        parent_id: str | None,
+        thread_id: str | None,
+    ) -> bool:
+        if chat_type == "p2p":
+            return False
+        return bool(root_id or parent_id or thread_id)
+
+    @staticmethod
+    def _compact_source_preview(text: str) -> str | None:
+        stripped = text.strip()
+        if not stripped:
+            return None
+
+        lines = [line.strip() for line in stripped.splitlines() if line.strip()]
+        if not lines:
+            return None
+        preview = "\n".join(lines[:3])
+        if len(preview) > 240:
+            preview = preview[:237].rstrip() + "..."
+        return preview
+
+    @classmethod
+    def _compose_card_text(cls, text: str, metadata: dict[str, Any] | None = None) -> str:
+        preview = None
+        if isinstance(metadata, dict):
+            raw_preview = metadata.get(SOURCE_PREVIEW_METADATA_KEY)
+            if isinstance(raw_preview, str) and raw_preview.strip():
+                preview = raw_preview.strip()
+        if not preview:
+            return text
+
+        quoted_preview = "\n".join(f"> {line}" for line in preview.splitlines())
+        return f"{quoted_preview}\n\n{text}"
+
     @property
     def supports_streaming(self) -> bool:
         return True
@@ -494,9 +534,15 @@ class FeishuChannel(Channel):
         self._background_tasks.discard(task)
         self._log_task_error(task, name, msg_id)
 
-    async def _create_running_card(self, source_message_id: str, text: str) -> str | None:
+    async def _create_running_card(
+        self,
+        source_message_id: str,
+        text: str,
+        *,
+        metadata: dict[str, Any] | None = None,
+    ) -> str | None:
         """Create the running card and cache its message ID when available."""
-        running_card_id = await self._reply_card(source_message_id, text)
+        running_card_id = await self._reply_card(source_message_id, self._compose_card_text(text, metadata))
         if running_card_id:
             self._running_card_ids[source_message_id] = running_card_id
             logger.info("[Feishu] running card created: source=%s card=%s", source_message_id, running_card_id)
@@ -504,7 +550,13 @@ class FeishuChannel(Channel):
             logger.warning("[Feishu] running card creation returned no message_id for source=%s, subsequent updates will fall back to new replies", source_message_id)
         return running_card_id
 
-    def _ensure_running_card_started(self, source_message_id: str, text: str = "thinking...") -> asyncio.Task | None:
+    def _ensure_running_card_started(
+        self,
+        source_message_id: str,
+        text: str = "thinking...",
+        *,
+        metadata: dict[str, Any] | None = None,
+    ) -> asyncio.Task | None:
         """Start running-card creation once per source message."""
         running_card_id = self._running_card_ids.get(source_message_id)
         if running_card_id:
@@ -514,7 +566,7 @@ class FeishuChannel(Channel):
         if running_card_task:
             return running_card_task
 
-        running_card_task = asyncio.create_task(self._create_running_card(source_message_id, text))
+        running_card_task = asyncio.create_task(self._create_running_card(source_message_id, text, metadata=metadata))
         self._running_card_tasks[source_message_id] = running_card_task
         running_card_task.add_done_callback(lambda done_task, mid=source_message_id: self._finalize_running_card_task(mid, done_task))
         return running_card_task
@@ -524,21 +576,31 @@ class FeishuChannel(Channel):
             self._running_card_tasks.pop(source_message_id, None)
         self._log_task_error(task, "create_running_card", source_message_id)
 
-    async def _ensure_running_card(self, source_message_id: str, text: str = "thinking...") -> str | None:
+    async def _ensure_running_card(
+        self,
+        source_message_id: str,
+        text: str = "thinking...",
+        *,
+        metadata: dict[str, Any] | None = None,
+    ) -> str | None:
         """Ensure the in-thread running card exists and track its message ID."""
         running_card_id = self._running_card_ids.get(source_message_id)
         if running_card_id:
             return running_card_id
 
-        running_card_task = self._ensure_running_card_started(source_message_id, text)
+        running_card_task = self._ensure_running_card_started(
+            source_message_id,
+            text,
+            metadata=metadata,
+        )
         if running_card_task is None:
             return self._running_card_ids.get(source_message_id)
         return await running_card_task
 
-    async def _send_running_reply(self, message_id: str) -> None:
+    async def _send_running_reply(self, message_id: str, *, metadata: dict[str, Any] | None = None) -> None:
         """Reply to a message in-thread with a running card."""
         try:
-            await self._ensure_running_card(message_id)
+            await self._ensure_running_card(message_id, metadata=metadata)
         except Exception:
             logger.exception("[Feishu] failed to send running reply for message %s", message_id)
 
@@ -556,8 +618,9 @@ class FeishuChannel(Channel):
                     running_card_id = await running_card_task
 
             if running_card_id:
+                card_text = self._compose_card_text(msg.text, msg.metadata)
                 try:
-                    await self._update_card(running_card_id, msg.text)
+                    await self._update_card(running_card_id, card_text)
                 except Exception:
                     if not msg.is_final:
                         raise
@@ -565,7 +628,7 @@ class FeishuChannel(Channel):
                         "[Feishu] failed to patch running card %s, falling back to final reply",
                         running_card_id,
                     )
-                    fallback_card_id = await self._reply_card(source_message_id, msg.text)
+                    fallback_card_id = await self._reply_card(source_message_id, card_text)
                     self._remember_thread_mapping(msg, source_message_id, fallback_card_id)
                     self._remember_pending_clarification(msg, fallback_card_id)
                 else:
@@ -573,7 +636,10 @@ class FeishuChannel(Channel):
                     self._remember_pending_clarification(msg, running_card_id)
                     logger.info("[Feishu] running card updated: source=%s card=%s", source_message_id, running_card_id)
             elif msg.is_final:
-                final_card_id = await self._reply_card(source_message_id, msg.text)
+                final_card_id = await self._reply_card(
+                    source_message_id,
+                    self._compose_card_text(msg.text, msg.metadata),
+                )
                 self._remember_thread_mapping(msg, source_message_id, final_card_id)
                 self._remember_pending_clarification(msg, final_card_id)
             elif awaited_running_card_task:
@@ -582,7 +648,11 @@ class FeishuChannel(Channel):
                     source_message_id,
                 )
             else:
-                created_card_id = await self._ensure_running_card(source_message_id, msg.text)
+                created_card_id = await self._ensure_running_card(
+                    source_message_id,
+                    msg.text,
+                    metadata=msg.metadata,
+                )
                 self._remember_thread_mapping(msg, source_message_id, created_card_id)
 
             if msg.is_final:
@@ -843,7 +913,7 @@ class FeishuChannel(Channel):
         for reaction_message_id in reaction_message_ids:
             reaction_task = asyncio.create_task(self._add_reaction(reaction_message_id, "OK"))
             self._track_background_task(reaction_task, name="add_reaction", msg_id=reaction_message_id)
-        self._ensure_running_card_started(msg_id)
+        self._ensure_running_card_started(msg_id, metadata=inbound.metadata)
         await self.bus.publish_inbound(inbound)
 
     async def _attach_connection_identity(self, inbound: InboundMessage) -> InboundMessage:
@@ -1014,6 +1084,27 @@ class FeishuChannel(Channel):
                     self._ensure_pending_thread_mapping(chat_id, sender_id, pending)
                     resolved_from_pending = True
 
+            source_preview = None
+            if self._should_include_source_preview(
+                chat_type=chat_type,
+                root_id=root_id,
+                parent_id=parent_id,
+                thread_id=feishu_thread_id,
+            ):
+                source_preview = self._compact_source_preview(text)
+
+            metadata = {
+                "message_id": msg_id,
+                "root_id": root_id,
+                "parent_id": parent_id,
+                "thread_id": feishu_thread_id,
+                "topic_id": topic_id,
+                "user_id": sender_id,
+                RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY: resolved_from_pending,
+            }
+            if source_preview:
+                metadata[SOURCE_PREVIEW_METADATA_KEY] = source_preview
+
             inbound = self._make_inbound(
                 chat_id=chat_id,
                 user_id=sender_id,
@@ -1021,15 +1112,7 @@ class FeishuChannel(Channel):
                 msg_type=msg_type,
                 thread_ts=msg_id,
                 files=files_list,
-                metadata={
-                    "message_id": msg_id,
-                    "root_id": root_id,
-                    "parent_id": parent_id,
-                    "thread_id": feishu_thread_id,
-                    "topic_id": topic_id,
-                    "user_id": sender_id,
-                    RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY: resolved_from_pending,
-                },
+                metadata=metadata,
             )
             inbound.topic_id = topic_id
 

backend/app/channels/feishu_run_policy.py

@@ -0,0 +1,15 @@
+"""Per-run policy registration for the Feishu channel."""
+
+from __future__ import annotations
+
+from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy
+
+
+def register_policy() -> None:
+    """Register Feishu's queue-same-thread behavior in the shared policy map."""
+    CHANNEL_RUN_POLICY["feishu"] = ChannelRunPolicy(
+        serialize_thread_runs=True,
+    )
+
+
+register_policy()

backend/app/channels/manager.py

@@ -17,6 +17,7 @@ from urllib.parse import quote
 import httpx
 from langgraph_sdk.errors import ConflictError
 
+from app.channels import feishu_run_policy as _feishu_run_policy  # noqa: F401
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
 from app.channels.message_bus import (
     PENDING_CLARIFICATION_METADATA_KEY,
@@ -29,6 +30,10 @@ from app.channels.message_bus import (
 from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy
 from app.channels.store import ChannelStore
 from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token
+
+# Import built-in channel run-policy registrars eagerly so direct
+# ChannelManager construction sees the same policy map as gateway bootstrap.
+from app.gateway.github import run_policy as _github_run_policy  # noqa: F401
 from app.gateway.internal_auth import create_internal_auth_headers
 from deerflow.config.agents_config import load_agent_config
 from deerflow.config.paths import make_safe_user_id
@@ -177,6 +182,14 @@ class _BoundIdentityRejection:
     outbound_owner_user_id: str | None = None
 
 
+@dataclass(slots=True)
+class _SerializedThreadRunState:
+    """Per-thread lock state for channels that queue same-thread turns."""
+
+    lock: asyncio.Lock
+    waiters: int = 0
+
+
 def _is_thread_busy_error(exc: BaseException | None) -> bool:
     if exc is None:
         return False
@@ -814,6 +827,9 @@ class ChannelManager:
         # Per-conversation locks so concurrent inbound messages for the same
         # chat don't race to create duplicate threads (see _get_or_create_thread).
         self._thread_create_locks: dict[tuple[str, str, str | None], asyncio.Lock] = {}
+        # Per-thread run locks for channels that want in-manager serialization
+        # instead of surfacing the runtime's generic busy reply.
+        self._serialized_thread_runs: dict[tuple[str, str], _SerializedThreadRunState] = {}
         self._skill_storage: SkillStorage | None = None
         self._csrf_token = generate_csrf_token()
         self._semaphore: asyncio.Semaphore | None = None
@@ -841,6 +857,57 @@ class ChannelManager:
         user_layer = _as_dict(users_layer.get(msg.user_id))
         return channel_layer, user_layer
 
+    def _begin_serialized_thread_run(
+        self,
+        *,
+        channel_name: str,
+        thread_id: str,
+    ) -> tuple[_SerializedThreadRunState | None, bool]:
+        policy = CHANNEL_RUN_POLICY.get(channel_name)
+        if policy is None or not policy.serialize_thread_runs:
+            return None, False
+
+        key = (channel_name, thread_id)
+        state = self._serialized_thread_runs.get(key)
+        if state is None:
+            state = _SerializedThreadRunState(lock=asyncio.Lock())
+            self._serialized_thread_runs[key] = state
+        queued = state.lock.locked()
+        state.waiters += 1
+        return state, queued
+
+    def _finish_serialized_thread_run(
+        self,
+        *,
+        channel_name: str,
+        thread_id: str,
+        state: _SerializedThreadRunState | None,
+        lock_acquired: bool,
+    ) -> None:
+        if state is None:
+            return
+
+        if lock_acquired:
+            state.lock.release()
+        state.waiters -= 1
+        if state.waiters == 0 and not state.lock.locked():
+            self._serialized_thread_runs.pop((channel_name, thread_id), None)
+
+    async def _publish_progress_update(self, msg: InboundMessage, thread_id: str, text: str) -> None:
+        await self.bus.publish_outbound(
+            OutboundMessage(
+                channel_name=msg.channel_name,
+                chat_id=msg.chat_id,
+                thread_id=thread_id,
+                text=text,
+                is_final=False,
+                thread_ts=msg.thread_ts,
+                connection_id=msg.connection_id,
+                owner_user_id=msg.owner_user_id,
+                metadata=_response_metadata(msg.metadata),
+            )
+        )
+
     def _resolve_run_params(self, msg: InboundMessage, thread_id: str) -> tuple[str, dict[str, Any], dict[str, Any]]:
         channel_layer, user_layer = self._resolve_session_layer(msg)
 
@@ -1439,6 +1506,50 @@ class ChannelManager:
             logger.info("[Manager] reusing thread: thread_id=%s for topic_id=%s", thread_id, msg.topic_id)
             await self._update_thread_channel_metadata(client, msg, thread_id)
 
+        serial_state, queued = self._begin_serialized_thread_run(
+            channel_name=msg.channel_name,
+            thread_id=thread_id,
+        )
+        serial_lock_acquired = False
+        try:
+            if queued:
+                await self._publish_progress_update(
+                    msg,
+                    thread_id,
+                    "Queued behind another request in this conversation. I’ll start working on this as soon as it finishes.",
+                )
+            if serial_state is not None:
+                await serial_state.lock.acquire()
+                serial_lock_acquired = True
+            if queued:
+                await self._publish_progress_update(msg, thread_id, "thinking...")
+            await self._handle_chat_on_thread(
+                client,
+                msg,
+                thread_id,
+                extra_context=extra_context,
+                storage_user_id=storage_user_id,
+            )
+        finally:
+            self._finish_serialized_thread_run(
+                channel_name=msg.channel_name,
+                thread_id=thread_id,
+                state=serial_state,
+                lock_acquired=serial_lock_acquired,
+            )
+
+    async def _handle_chat_on_thread(
+        self,
+        client,
+        msg: InboundMessage,
+        thread_id: str,
+        *,
+        extra_context: dict[str, Any] | None = None,
+        storage_user_id: str | None = None,
+    ) -> None:
+        if storage_user_id is None:
+            storage_user_id = _channel_storage_user_id(msg)
+
         assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id)
 
         # Apply per-channel policy: credentials provider (e.g. GitHub

backend/app/channels/run_policy.py

@@ -76,6 +76,14 @@ class ChannelRunPolicy:
             fires. Defaults to False (the safe default for an
             interactive IM channel that depends on the manager to
             publish the agent's reply).
+        serialize_thread_runs: When True, the manager serializes
+            same-thread inbound turns for this channel instead of
+            surfacing the runtime's generic busy-thread error. This is
+            useful for chat surfaces like Feishu topics where rapid
+            follow-up messages should queue behind the active turn while
+            unrelated DeerFlow threads continue concurrently. Defaults
+            to False so existing channels keep the runtime's native
+            multitask behavior unless they opt in explicitly.
     """
 
     is_interactive: bool = True
@@ -83,6 +91,7 @@ class ChannelRunPolicy:
     credentials_provider: Callable[[InboundMessage, dict[str, Any]], Awaitable[None]] | None = None
     requires_bound_identity: bool = True
     fire_and_forget: bool = False
+    serialize_thread_runs: bool = False
 
 
 # Channel name → policy. Channels absent from this map fall through to

backend/tests/test_channels.py

@@ -1816,6 +1816,232 @@ class TestChannelManager:
 
         _run(go())
 
+    def test_handle_feishu_same_thread_messages_queue_instead_of_busy(self, monkeypatch):
+        from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
+
+        monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            outbound_received = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            first_started = asyncio.Event()
+            release_first = asyncio.Event()
+            second_started = asyncio.Event()
+
+            async def _stream(thread_id, assistant_id, *, input, **kwargs):  # noqa: ARG001
+                prompt = input["messages"][0]["content"]
+                if prompt == "first":
+                    first_started.set()
+                    await release_first.wait()
+                    yield _make_stream_part(
+                        "values",
+                        {
+                            "messages": [
+                                {"type": "human", "content": "first"},
+                                {"type": "ai", "content": "First done"},
+                            ],
+                            "artifacts": [],
+                        },
+                    )
+                    return
+
+                second_started.set()
+                yield _make_stream_part(
+                    "values",
+                    {
+                        "messages": [
+                            {"type": "human", "content": "second"},
+                            {"type": "ai", "content": "Second done"},
+                        ],
+                        "artifacts": [],
+                    },
+                )
+
+            mock_client = _make_mock_langgraph_client(thread_id="feishu-thread-1")
+            mock_client.runs.stream = MagicMock(side_effect=_stream)
+            manager._client = mock_client
+
+            await manager.start()
+
+            await bus.publish_inbound(
+                InboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat1",
+                    user_id="user1",
+                    text="first",
+                    topic_id="topic-1",
+                    thread_ts="om-source-1",
+                )
+            )
+            await _wait_for(first_started.is_set)
+
+            await bus.publish_inbound(
+                InboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat1",
+                    user_id="user1",
+                    text="second",
+                    topic_id="topic-1",
+                    thread_ts="om-source-2",
+                )
+            )
+
+            await _wait_for(lambda: any(message.thread_ts == "om-source-2" and message.text.startswith("Queued behind another request") for message in outbound_received))
+            assert second_started.is_set() is False
+
+            release_first.set()
+            await _wait_for(second_started.is_set)
+            await _wait_for(lambda: len([message for message in outbound_received if message.is_final]) == 2)
+            await manager.stop()
+
+            assert all(message.text != THREAD_BUSY_MESSAGE for message in outbound_received)
+            second_turn = [message for message in outbound_received if message.thread_ts == "om-source-2"]
+            assert second_turn[0].text.startswith("Queued behind another request")
+            assert any(message.text == "thinking..." for message in second_turn if message.is_final is False)
+            assert second_turn[-1].text == "Second done"
+            assert mock_client.runs.stream.call_count == 2
+
+        _run(go())
+
+    def test_handle_feishu_queue_waiter_cleanup_on_cancelled_progress_publish(self):
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            msg = InboundMessage(
+                channel_name="feishu",
+                chat_id="chat1",
+                user_id="user1",
+                text="second",
+                topic_id="topic-1",
+                thread_ts="om-source-2",
+            )
+
+            thread_id = "feishu-thread-1"
+            serial_state, _ = manager._begin_serialized_thread_run(
+                channel_name="feishu",
+                thread_id=thread_id,
+            )
+            assert serial_state is not None
+            await serial_state.lock.acquire()
+
+            manager._get_client = MagicMock(return_value=object())
+            manager._get_or_create_thread = AsyncMock(return_value=(thread_id, False))
+            manager._update_thread_channel_metadata = AsyncMock()
+            manager._publish_progress_update = AsyncMock(side_effect=asyncio.CancelledError())
+            manager._handle_chat_on_thread = AsyncMock()
+
+            with pytest.raises(asyncio.CancelledError):
+                await manager._handle_chat(msg, bound_identity_checked=True)
+
+            leaked_state = manager._serialized_thread_runs.get(("feishu", thread_id))
+            assert leaked_state is serial_state
+            assert leaked_state.waiters == 1
+            assert leaked_state.lock.locked() is True
+            manager._handle_chat_on_thread.assert_not_awaited()
+
+            manager._finish_serialized_thread_run(
+                channel_name="feishu",
+                thread_id=thread_id,
+                state=serial_state,
+                lock_acquired=True,
+            )
+            assert ("feishu", thread_id) not in manager._serialized_thread_runs
+
+        _run(go())
+
+    def test_handle_feishu_different_threads_can_stream_concurrently(self, monkeypatch):
+        from app.channels.manager import ChannelManager
+
+        monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            first_started = asyncio.Event()
+            second_started = asyncio.Event()
+            release_streams = asyncio.Event()
+
+            async def create_thread(**kwargs):
+                topic_id = kwargs["metadata"]["channel_source"]["topic_id"]
+                return {"thread_id": f"thread-{topic_id}"}
+
+            async def _stream(thread_id, assistant_id, *, input, **kwargs):  # noqa: ARG001
+                if thread_id == "thread-topic-a":
+                    first_started.set()
+                elif thread_id == "thread-topic-b":
+                    second_started.set()
+                await release_streams.wait()
+                yield _make_stream_part(
+                    "values",
+                    {
+                        "messages": [
+                            {"type": "human", "content": input["messages"][0]["content"]},
+                            {"type": "ai", "content": f"done:{thread_id}"},
+                        ],
+                        "artifacts": [],
+                    },
+                )
+
+            mock_client = _make_mock_langgraph_client()
+            mock_client.threads.create = AsyncMock(side_effect=create_thread)
+            mock_client.runs.stream = MagicMock(side_effect=_stream)
+            manager._client = mock_client
+
+            outbound_received = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            await manager.start()
+            await bus.publish_inbound(
+                InboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat1",
+                    user_id="user1",
+                    text="first",
+                    topic_id="topic-a",
+                    thread_ts="om-source-a",
+                )
+            )
+            await bus.publish_inbound(
+                InboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat1",
+                    user_id="user1",
+                    text="second",
+                    topic_id="topic-b",
+                    thread_ts="om-source-b",
+                )
+            )
+
+            await _wait_for(first_started.is_set)
+            await _wait_for(second_started.is_set)
+            release_streams.set()
+            await _wait_for(lambda: len([message for message in outbound_received if message.is_final]) == 2)
+            await manager.stop()
+
+            assert mock_client.runs.stream.call_count == 2
+            assert not any(message.text.startswith("Queued behind another request") for message in outbound_received)
+
+        _run(go())
+
     def test_handle_command_help(self):
         from app.channels.manager import ChannelManager
 
@@ -3123,6 +3349,16 @@ class TestGithubFireAndForget:
         from app.channels.run_policy import ChannelRunPolicy
 
         assert ChannelRunPolicy().fire_and_forget is False
+        assert ChannelRunPolicy().serialize_thread_runs is False
+
+    def test_feishu_channel_policy_opts_into_serialized_thread_runs(self):
+        """Feishu's queue-same-thread behavior should be policy-driven."""
+        import app.channels.feishu_run_policy  # noqa: F401
+        from app.channels.run_policy import CHANNEL_RUN_POLICY
+
+        feishu_policy = CHANNEL_RUN_POLICY.get("feishu")
+        assert feishu_policy is not None
+        assert feishu_policy.serialize_thread_runs is True
 
     def test_github_channel_policy_opts_into_fire_and_forget(self):
         """The GitHub channel must register ``fire_and_forget=True``. This is
@@ -4358,6 +4594,48 @@ class TestFeishuChannel:
 
         _run(go())
 
+    def test_prepare_inbound_topic_reply_includes_source_preview(self):
+        from app.channels.feishu import SOURCE_PREVIEW_METADATA_KEY, FeishuChannel
+
+        async def go():
+            bus = MessageBus()
+            bus.publish_inbound = AsyncMock()
+            channel = FeishuChannel(bus, config={})
+
+            reply_started = asyncio.Event()
+            release_reply = asyncio.Event()
+
+            async def slow_reply(message_id: str, text: str) -> str:
+                reply_started.set()
+                await release_reply.wait()
+                return "om-running-card"
+
+            channel._add_reaction = AsyncMock()
+            channel._reply_card = AsyncMock(side_effect=slow_reply)
+
+            inbound = InboundMessage(
+                channel_name="feishu",
+                chat_id="chat-1",
+                user_id="user-1",
+                text="follow-up question",
+                thread_ts="om-source-msg",
+                metadata={SOURCE_PREVIEW_METADATA_KEY: "follow-up question"},
+            )
+
+            prepare_task = asyncio.create_task(channel._prepare_inbound("om-source-msg", inbound))
+
+            await _wait_for(lambda: bus.publish_inbound.await_count == 1)
+            await _wait_for(reply_started.is_set)
+
+            preview_text = channel._reply_card.await_args.args[1]
+            assert preview_text == "> follow-up question\n\nthinking..."
+
+            await prepare_task
+            release_reply.set()
+            await _wait_for(lambda: channel._running_card_ids.get("om-source-msg") == "om-running-card")
+
+        _run(go())
+
     def test_prepare_inbound_and_send_share_running_card_task(self):
         from app.channels.feishu import FeishuChannel
 
@@ -4497,6 +4775,74 @@ class TestFeishuChannel:
 
         _run(go())
 
+    def test_streaming_updates_preserve_source_preview(self):
+        from lark_oapi.api.im.v1 import (
+            CreateMessageReactionRequest,
+            CreateMessageReactionRequestBody,
+            Emoji,
+            PatchMessageRequest,
+            PatchMessageRequestBody,
+            ReplyMessageRequest,
+            ReplyMessageRequestBody,
+        )
+
+        from app.channels.feishu import SOURCE_PREVIEW_METADATA_KEY, FeishuChannel
+
+        async def go():
+            bus = MessageBus()
+            channel = FeishuChannel(bus, config={})
+
+            channel._api_client = MagicMock()
+            channel._ReplyMessageRequest = ReplyMessageRequest
+            channel._ReplyMessageRequestBody = ReplyMessageRequestBody
+            channel._PatchMessageRequest = PatchMessageRequest
+            channel._PatchMessageRequestBody = PatchMessageRequestBody
+            channel._CreateMessageReactionRequest = CreateMessageReactionRequest
+            channel._CreateMessageReactionRequestBody = CreateMessageReactionRequestBody
+            channel._Emoji = Emoji
+
+            reply_response = MagicMock()
+            reply_response.data.message_id = "om-running-card"
+            channel._api_client.im.v1.message.reply = MagicMock(return_value=reply_response)
+            channel._api_client.im.v1.message.patch = MagicMock()
+            channel._api_client.im.v1.message_reaction.create = MagicMock()
+
+            metadata = {SOURCE_PREVIEW_METADATA_KEY: "What changed in the last run?"}
+
+            await channel._send_running_reply("om-source-msg", metadata=metadata)
+            await channel.send(
+                OutboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat-1",
+                    thread_id="thread-1",
+                    text="Queued behind another request",
+                    is_final=False,
+                    thread_ts="om-source-msg",
+                    metadata=metadata,
+                )
+            )
+            await channel.send(
+                OutboundMessage(
+                    channel_name="feishu",
+                    chat_id="chat-1",
+                    thread_id="thread-1",
+                    text="Answer ready",
+                    is_final=True,
+                    thread_ts="om-source-msg",
+                    metadata=metadata,
+                )
+            )
+
+            reply_request = channel._api_client.im.v1.message.reply.call_args.args[0]
+            first_patch_request = channel._api_client.im.v1.message.patch.call_args_list[0].args[0]
+            final_patch_request = channel._api_client.im.v1.message.patch.call_args_list[1].args[0]
+
+            assert json.loads(reply_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nthinking..."
+            assert json.loads(first_patch_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nQueued behind another request"
+            assert json.loads(final_patch_request.body.content)["elements"][0]["content"] == "> What changed in the last run?\n\nAnswer ready"
+
+        _run(go())
+
 
 class TestWeComChannel:
     def test_publish_ws_inbound_starts_stream_and_publishes_message(self, monkeypatch):

backend/tests/test_feishu_parser.py

@@ -328,7 +328,13 @@ def test_feishu_batches_top_level_file_messages_from_same_user(monkeypatch):
         assert inbound.metadata["message_id"] == "msg_file_1"
         assert inbound.metadata["topic_id"] == "msg_file_1"
         assert inbound.metadata["batched_message_ids"] == ["msg_file_1", "msg_file_2"]
-        channel._ensure_running_card_started.assert_called_once_with("msg_file_1")
+        channel._ensure_running_card_started.assert_called_once()
+        assert channel._ensure_running_card_started.call_args.args == ("msg_file_1",)
+        assert channel._ensure_running_card_started.call_args.kwargs["metadata"]["message_id"] == "msg_file_1"
+        assert channel._ensure_running_card_started.call_args.kwargs["metadata"]["batched_message_ids"] == [
+            "msg_file_1",
+            "msg_file_2",
+        ]
         assert [call.args for call in channel._add_reaction.call_args_list] == [
             ("msg_file_1", "OK"),
             ("msg_file_2", "OK"),
@@ -390,6 +396,8 @@ def test_feishu_file_batch_window_expiry_starts_new_topic(monkeypatch):
         assert second.files == [{"file_key": "file_b"}]
         assert channel._ensure_running_card_started.call_args_list[0].args == ("msg_file_1",)
         assert channel._ensure_running_card_started.call_args_list[1].args == ("msg_file_2",)
+        assert channel._ensure_running_card_started.call_args_list[0].kwargs["metadata"]["message_id"] == "msg_file_1"
+        assert channel._ensure_running_card_started.call_args_list[1].kwargs["metadata"]["message_id"] == "msg_file_2"
 
     _run(go())