fix(channels): keep IM runs alive when the creating request disconnects (#6876)

* fix(channels): keep IM runs alive when the creating request disconnects

The channel manager holds a `runs.wait` / `runs.stream` request open for
the whole run, and that request dies with the Gateway instance that
received the IM message: a leader hand-off, a rolling restart, a crash.
The Gateway cancels a run whose creating request disconnects unless the
request asked for `on_disconnect: continue`, so every IM run in flight
was cancelled on failover.

Channel runs now ask for `continue`, like scheduled-task and MCP-task
runs already do. The run finishes on the Gateway executing it and its
thread state is persisted, so the answer stays readable in the web UI
even though the IM reply is lost with the old transport. The key is set
after the fire-and-forget branch because the SDK's `runs.create` has no
`on_disconnect` parameter.

Docs: CONFIGURATION.md and the Helm/deployment guides no longer list
"failover cancels IM runs" as a known limit; IM_CHANNEL_CONNECTIONS.md
and the channels AGENTS.md describe the policy.

* docs(changelog): link #6876

* fix(channels): keep the inbound dedupe key once a run may be executing

With `on_disconnect: continue` a run outlives the manager worker that
requested it, but the worker loop released the inbound dedupe key on
cancellation, so a provider redelivery on the replacement instance
admitted the same event again and executed the turn twice (review on
#6876).

Each inbound message now carries an admission record: `requested`
flips before the run request leaves the manager, `confirmed` once the
Gateway proved it admitted the run (a run id, a stream frame, a
completed response). The generic release paths (worker cancellation,
worker/handler exceptions, a broken stream) keep the key when the run
may continue: after confirmation, on cancellation after the request
was sent, and on post-request transport failures (read timeout, read
error, protocol error). Pre-request failures, Gateway rejections
(busy thread, stale pin, status errors, connection failures) and the
stream's terminal `error` frame still release it so the redelivery
can retry.

CHANGELOG.md

@@ -1247,6 +1247,26 @@ This release closes that milestone with **621 merged pull requests**.
 
 ### Fixed
 
+- **channels:** Runs started from an IM channel no longer die with the Gateway
+  instance that received the message. The manager holds `runs.wait` /
+  `runs.stream` open for the whole run, and the Gateway cancels a run whose
+  creating request disconnects unless that request asked for
+  `on_disconnect: continue`; a leader hand-off, a rolling restart or a crash
+  of the receiving instance therefore cancelled every IM run in flight.
+  Channel runs now ask for `continue`, so the run finishes on the Gateway
+  executing it and its thread state is persisted — the answer stays readable
+  in the web UI even though the IM reply is lost with the old transport. The
+  same policy covers the SDK's 300s read timeout on a Slack/Discord
+  `runs.wait`: the chat still gets the error reply, but the run finishes in
+  the background instead of being cancelled. Because the run survives, the
+  manager now keeps the inbound dedupe key once the run request has left it
+  (a cancelled worker, a read timeout, a dropped connection): the provider's
+  redelivery on the new holder is dropped instead of executing the turn a
+  second time, while a request the Gateway rejected or a failure before
+  sending still releases the key for a retry. Fire-and-forget channels
+  (GitHub) are unchanged: `runs.create` returns as soon as the run is pending.
+  ([#6876])
+
 - **deploy:** The Gateway container now shuts down gracefully. The production
   compose `command`, both `CMD`s in `backend/Dockerfile` and the Helm chart
   started it as `sh -c "cd backend && … uv run … uvicorn …"`, which leaves `sh`
@@ -11711,3 +11731,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6844]: https://github.com/bytedance/deer-flow/pull/6844
 [#6852]: https://github.com/bytedance/deer-flow/pull/6852
 [#6865]: https://github.com/bytedance/deer-flow/pull/6865
+[#6876]: https://github.com/bytedance/deer-flow/pull/6876

CHANGELOG_zh.md

@@ -1002,6 +1002,18 @@
 
 ### 修复
 
+- **渠道：** 从 IM 渠道发起的运行不再随接收消息的 Gateway 实例一起消亡。manager 在整个
+  运行期间保持 `runs.wait` / `runs.stream` 请求打开，而 Gateway 会在创建请求断开时取消
+  运行，除非该请求声明了 `on_disconnect: continue`；因此 leader 切换、滚动重启或接收
+  实例崩溃都会取消所有进行中的 IM 运行。渠道运行现在声明 `continue`，运行会在执行它的
+  Gateway 上完成并持久化对话线程状态——即使 IM 回复随旧传输丢失，答案仍可在 Web UI
+  中查看。同一策略也覆盖 Slack/Discord `runs.wait` 遇到 SDK 300 秒读超时的情况：聊天里
+  仍会收到错误回复，但运行会在后台完成而不再被取消。由于运行会继续，manager 现在在运行
+  请求发出之后保留入站去重键（worker 被取消、读超时、连接断开）：provider 在新持有者上的
+  重投递会被丢弃，而不是把这一轮再执行一遍；Gateway 拒绝的请求或发送前的失败仍会释放该键
+  以便重试。即发即弃渠道（GitHub）不受影响：`runs.create` 在运行进入 pending 后即返回。
+  ([#6876])
+
 - **前端：** 用户管理设置页不再在静态网站演示中发出注定失败的查询，非管理员的
   `?settings=users` 深链接也不再打开仅管理员可见的面板。该页面此前只以
   `system_role === "admin"` 判断，而演示用户名义上是管理员，静态响应层又没有
@@ -9345,3 +9357,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6844]: https://github.com/bytedance/deer-flow/pull/6844
 [#6852]: https://github.com/bytedance/deer-flow/pull/6852
 [#6865]: https://github.com/bytedance/deer-flow/pull/6865
+[#6876]: https://github.com/bytedance/deer-flow/pull/6876

backend/app/channels/manager.py

@@ -11,7 +11,9 @@ import stat
 import time
 import uuid
 from collections import OrderedDict
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass
 from html import escape
 from pathlib import Path
@@ -100,6 +102,74 @@ STREAM_UPDATE_MIN_CHARS = 60  # flush immediately when this many chars accumulat
 STREAM_MODES = ["messages-tuple", "values"]
 MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages")
 THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."
+
+# IM runs must outlive the request that created them. The manager holds a
+# ``runs.wait`` / ``runs.stream`` request open for the whole run, and that
+# request dies with this Gateway instance: a leader hand-off, a rolling
+# restart, a crash. The Gateway's default ``on_disconnect="cancel"`` would then
+# cancel the run itself; ``continue`` lets it finish and persist its thread
+# state, so the user still finds the answer in the Web UI even though the IM
+# reply is lost with the transport. ``runs.create`` (fire-and-forget) needs no
+# such policy: its request returns as soon as the run is pending.
+CHANNEL_RUN_ON_DISCONNECT = "continue"
+
+
+@dataclass
+class _RunAdmission:
+    """How far one inbound message's run request got, for dedupe release.
+
+    ``requested`` flips right before the run request leaves the manager;
+    ``confirmed`` once the Gateway proved it admitted the run (a run id, a
+    stream frame, a completed response). See ``_keep_dedupe_key_after``.
+    """
+
+    requested: bool = False
+    confirmed: bool = False
+
+
+_RUN_ADMISSION: ContextVar[_RunAdmission | None] = ContextVar("deerflow_channel_run_admission", default=None)
+
+# Transport failures that can only happen after the Gateway read the request:
+# the run was admitted and continues (``on_disconnect: continue``), so a
+# provider redelivery must stay a duplicate. Connection failures and HTTP
+# status errors mean the Gateway never admitted it.
+_RUN_REQUEST_REACHED_GATEWAY_ERRORS = (httpx.ReadTimeout, httpx.ReadError, httpx.RemoteProtocolError)
+
+
+class _RunFailedFrame(RuntimeError):
+    """The stream's ``error`` frame: the run ended, so a retry is not a duplicate."""
+
+
+@contextmanager
+def _run_admission_scope(*, fresh: bool = False) -> Iterator[_RunAdmission]:
+    """Install the admission record for one inbound message.
+
+    The worker loop installs a fresh one per message; direct callers of
+    ``_handle_message`` get one only when none is installed yet.
+    """
+    existing = _RUN_ADMISSION.get()
+    if existing is not None and not fresh:
+        yield existing
+        return
+    admission = _RunAdmission()
+    token = _RUN_ADMISSION.set(admission)
+    try:
+        yield admission
+    finally:
+        _RUN_ADMISSION.reset(token)
+
+
+def _note_run_requested() -> None:
+    if (admission := _RUN_ADMISSION.get()) is not None:
+        admission.requested = True
+
+
+def _note_run_confirmed() -> None:
+    if (admission := _RUN_ADMISSION.get()) is not None:
+        admission.requested = True
+        admission.confirmed = True
+
+
 # Recovery replies must say the turn was dropped: the run was rejected before
 # starting, so the triggering message never reached the agent.
 _STALE_PIN_RETRY_NOTE = "Your message was not processed — please send it again."
@@ -2261,7 +2331,7 @@ class ChannelManager:
             # for them. Scope one trace id per message here -- the worker task
             # is long-lived and reused, so the scope must close with the
             # message rather than leak into the next one.
-            with ensure_trace_context():
+            with ensure_trace_context(), _run_admission_scope(fresh=True):
                 dedupe_recorded = False
                 try:
                     # Dedupe before logging "received" so a provider retrying an
@@ -2280,24 +2350,25 @@ class ChannelManager:
                     )
                     # Deliberately awaited inline: never create a task per message.
                     await self._handle_message(msg)
-                except asyncio.CancelledError:
+                except asyncio.CancelledError as exc:
                     # A cancellation after dedupe admission must make provider
                     # redelivery retryable rather than retaining a TTL-long key for
-                    # work that never completed.
-                    if dedupe_recorded:
+                    # work that never completed -- unless the Gateway may already
+                    # be executing the run, which outlives this worker.
+                    if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc):
                         try:
                             await self._release_inbound_dedupe_key(msg)
                         except Exception:
                             logger.exception("[Manager] failed to release inbound dedupe key during worker cancellation")
                     raise
-                except Exception:
+                except Exception as exc:
                     logger.exception(
                         "[Manager] inbound worker %d failed handling channel=%s chat_id=%s",
                         worker_index,
                         msg.channel_name,
                         msg.chat_id,
                     )
-                    if dedupe_recorded:
+                    if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc):
                         try:
                             await self._release_inbound_dedupe_key(msg)
                         except Exception:
@@ -2368,11 +2439,43 @@ class ChannelManager:
         recorded on receipt so retries arriving *while* the message is being
         handled are still deduped, but if handling fails we must not turn a
         recoverable error into a TTL-long black hole for the same message_id.
+        Generic failure paths ask ``_keep_dedupe_key_after`` first; sites that
+        know the Gateway rejected the run (busy thread, stale pin) release
+        unconditionally.
         """
         key = self._inbound_dedupe_key(msg)
         if key is not None:
             await self._inbound_dedupe_store.release(key)
 
+    @staticmethod
+    def _keep_dedupe_key_after(msg: InboundMessage, exc: BaseException | None) -> bool:
+        """Whether *msg*'s dedupe key must outlive a failed or cancelled handling.
+
+        Runs are created with ``on_disconnect: continue``, so once the Gateway
+        may have admitted this message's run, the run outlives the request and
+        a provider redelivery must stay a duplicate: a second run would repeat
+        the first one's side effects. The key is released, so the redelivery
+        can retry, only while the manager still owns the outcome: before the
+        run request leaves it, when the Gateway provably rejected the request
+        (an HTTP status error, a connection that never opened), or after a
+        terminal ``error`` frame ended the run. ``/wait`` confirms admission
+        only with its final response, so a cancellation or a post-request
+        transport failure while it is in flight counts as admitted rather than
+        risking a duplicate run.
+        """
+        admission = _RUN_ADMISSION.get()
+        if admission is None or not admission.requested or isinstance(exc, _RunFailedFrame):
+            return False
+        if admission.confirmed or isinstance(exc, (asyncio.CancelledError, *_RUN_REQUEST_REACHED_GATEWAY_ERRORS)):
+            logger.info(
+                "[Manager] keeping inbound dedupe key, the run may continue on the Gateway: channel=%s, chat_id=%s, cause=%s",
+                msg.channel_name,
+                msg.chat_id,
+                type(exc).__name__ if exc is not None else "completed",
+            )
+            return True
+        return False
+
     @staticmethod
     def _log_task_error(task: asyncio.Task) -> None:
         """Surface unhandled exceptions from background tasks."""
@@ -2383,6 +2486,10 @@ class ChannelManager:
             logger.error("[Manager] unhandled error in message task: %s", exc, exc_info=exc)
 
     async def _handle_message(self, msg: InboundMessage) -> None:
+        with _run_admission_scope():
+            await self._dispatch_message(msg)
+
+    async def _dispatch_message(self, msg: InboundMessage) -> None:
         msg = _apply_effective_owner(msg)
         try:
             # Commands are handled below because provider adapters consume
@@ -2415,7 +2522,7 @@ class ChannelManager:
                 exc,
             )
             await self._send_error(msg, str(exc))
-        except Exception:
+        except Exception as exc:
             logger.exception(
                 "Error handling message from %s (chat=%s)",
                 msg.channel_name,
@@ -2423,9 +2530,11 @@ class ChannelManager:
             )
             # Transient/unexpected failure: release the dedupe key so a provider
             # redelivery of the same message can recover instead of being dropped
-            # for the dedupe TTL. Stale /model pin admission failures are handled
-            # at the run-creation sites, which know the selection the run used.
-            await self._release_inbound_dedupe_key(msg)
+            # for the dedupe TTL -- unless the run may already be executing.
+            # Stale /model pin admission failures are handled at the
+            # run-creation sites, which know the selection the run used.
+            if not self._keep_dedupe_key_after(msg, exc):
+                await self._release_inbound_dedupe_key(msg)
             await self._send_error(msg, "An internal error occurred. Please try again.")
 
     # -- chat handling -----------------------------------------------------
@@ -3005,6 +3114,7 @@ class ChannelManager:
                 thread_id,
                 len(msg.text or ""),
             )
+            _note_run_requested()
             try:
                 # Capturing the return value is new (issue #4121 Slice 2):
                 # it carries ``run_id``, which the follow-up watcher below
@@ -3029,11 +3139,15 @@ class ChannelManager:
                     await self._release_inbound_dedupe_key(msg)
                     return
                 raise
+            _note_run_confirmed()
             if policy.buffer_followups_on_busy:
                 self._maybe_spawn_followup_watcher(thread_id, result, msg)
             return
 
+        # Only now: ``runs.create`` above has no ``on_disconnect`` parameter.
+        run_kwargs["on_disconnect"] = CHANNEL_RUN_ON_DISCONNECT
         logger.info("[Manager] invoking runs.wait(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))
+        _note_run_requested()
         try:
             result = await client.runs.wait(
                 thread_id,
@@ -3053,6 +3167,7 @@ class ChannelManager:
                 await self._send_error(msg, recovery.text)
                 return
             raise
+        _note_run_confirmed()
 
         response_text = _extract_response_text(result)
         pending_clarification = _has_current_turn_clarification(result)
@@ -3121,17 +3236,21 @@ class ChannelManager:
             "context": run_context,
             "stream_mode": list(STREAM_MODES),
             "multitask_strategy": "reject",
+            "on_disconnect": CHANNEL_RUN_ON_DISCONNECT,
         }
         _apply_run_origin(stream_kwargs, msg)
         if owner_headers := _owner_headers(msg):
             stream_kwargs["headers"] = owner_headers
 
+        _note_run_requested()
         try:
             async for chunk in client.runs.stream(
                 thread_id,
                 assistant_id,
                 **stream_kwargs,
             ):
+                # Any frame proves the Gateway admitted the run.
+                _note_run_confirmed()
                 event = getattr(chunk, "event", "")
                 data = getattr(chunk, "data", None)
 
@@ -3149,7 +3268,7 @@ class ChannelManager:
                     )
                     # The v1 SDK yields backend errors as frames, not exceptions.
                     # Keep draining the stream and use the existing failure path.
-                    stream_error = RuntimeError("Agent run failed.")
+                    stream_error = _RunFailedFrame("Agent run failed.")
                 elif event == "values" and isinstance(data, (dict, list)):
                     last_values = data
                     # Clarification text is only in the values snapshot;
@@ -3241,11 +3360,12 @@ class ChannelManager:
                     metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification),
                 )
             )
-            if stream_error is not None:
+            if stream_error is not None and not self._keep_dedupe_key_after(msg, stream_error):
                 # This path swallows its own errors, so _handle_message's generic
                 # handler never runs and never releases the key. Release only
                 # after publishing the final outbound so a provider redelivery
-                # cannot overtake this attempt's terminal reply.
+                # cannot overtake this attempt's terminal reply. A stream that
+                # broke after the run was admitted keeps the key: the run goes on.
                 await self._release_inbound_dedupe_key(msg)
 
     # -- command handling --------------------------------------------------

backend/docs/CONFIGURATION.md

@@ -814,7 +814,7 @@ Every IM channel except GitHub connects outbound to its platform (Telegram long
 - Runtime channel configuration does not cross instances yet: `POST /api/channels/{provider}/runtime-config` saves the values (in the shared `runtime-config.json`) and applies them only where the provider's transport runs — a non-holder accepts them without restarting the holder, which keeps its current credentials until it next starts the channel (an instance taking a lease over reloads the provider's saved config first). Make credential changes on a single instance or restart the holder afterwards.
 - The block is validated on every topology: an unknown key, a non-mapping value, `enabled` outside `auto | true | false`, `lease_seconds < 3` or `grace_seconds < 0` stop startup with the dotted path in the message.
 - Restart-required (`channels.leader_lease` in the reload-boundary registry): the gate evaluates it once and `ChannelService` captures it at construction. Change it on every instance and restart them together.
-- Known limits: a leader change cancels the IM runs in flight on the old holder (their `on_disconnect` is `cancel`), and transport-local state such as Feishu running cards is lost on failover.
+- Known limits: a leader change loses the IM replies in flight on the old holder. The runs themselves continue on the Gateway executing them (channel runs are created with `on_disconnect: continue`) and their thread state is persisted, so the answer stays readable in the web UI, but the transport-local state that would deliver it (a Telegram placeholder message, a Feishu running card, a DingTalk replier) is gone with the old holder; a run executing on the instance that died is recovered as an error like any run of a dead worker.
 
 ### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`)
 

backend/docs/IM_CHANNEL_CONNECTIONS.md

@@ -631,7 +631,16 @@ platform connection to one instance: the holder of the provider's row in
 state, so the connect flow works on any instance; `GET /api/channels/` names
 each provider's `leader`. Runtime credentials saved on an instance that does
 not hold the provider apply on the holder only when it next starts the
-channel. See
+channel. Channel runs are created with `on_disconnect: continue`, so a run
+keeps executing when the request that created it dies with its instance (a
+leader hand-off, a rolling restart, a crash): its thread state is persisted
+and the answer stays readable in the web UI, while the IM reply itself is
+lost with the old holder's transport state. Because that run survives, the
+manager keeps the inbound dedupe key once a run request has left it (a
+cancelled worker, a read timeout, a dropped connection), so the provider's
+redelivery of the same message on the new holder is dropped instead of
+executing the turn twice; a request the Gateway rejected, or a failure before
+the request was sent, still releases the key for a retry. See
 [IM channel leader lease](CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease).
 
 ## Security Notes

backend/tests/test_channels.py

@@ -12,9 +12,11 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
 
 from app.channels.base import Channel
+from app.channels.dedupe_store import MemoryInboundDedupeStore
 from app.channels.message_bus import (
     PENDING_CLARIFICATION_METADATA_KEY,
     InboundMessage,
@@ -1435,15 +1437,22 @@ class TestChannelManager:
 
         _run(go())
 
-    def test_streaming_transient_failure_releases_dedupe_key(self, tmp_path, monkeypatch):
-        """Release a swallowed streaming error only after its final outbound.
+    @pytest.mark.parametrize("admitted", [False, True], ids=["before-admission", "after-admission"])
+    def test_streaming_transient_failure_releases_dedupe_key_only_before_admission(self, tmp_path, monkeypatch, admitted):
+        """A swallowed streaming error releases the key only while the manager
+        still owns the outcome, and only after its final outbound.
 
         _release_inbound_dedupe_key lives in _handle_message's `except Exception`
         handler, but _handle_streaming_chat handles its own errors and never
         re-raises — so without an explicit release the key recorded on receipt
         survives the full dedupe TTL and the provider's redelivery (the retry
-        that would recover the failure) is silently dropped. Releasing before
-        the final outbound would let that retry overtake the terminal reply.
+        that would recover a failure) is silently dropped. Releasing before the
+        final outbound would let that retry overtake the terminal reply.
+
+        Once the Gateway has admitted the run (a frame arrived), a broken
+        stream no longer releases the key: the run continues on the Gateway
+        (``on_disconnect: continue``), so the redelivery would execute the
+        same turn twice.
         """
         monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
         from app.channels.manager import ChannelManager
@@ -1465,10 +1474,11 @@ class TestChannelManager:
 
             def _failing_stream(*args, **kwargs):
                 async def gen():
-                    yield _make_stream_part(
-                        "messages-tuple",
-                        [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}],
-                    )
+                    if admitted:
+                        yield _make_stream_part(
+                            "messages-tuple",
+                            [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}],
+                        )
                     raise ConnectionError("stream broken")
 
                 return gen()
@@ -1494,10 +1504,12 @@ class TestChannelManager:
 
             # The provider redelivers the same message after the failure.
             await bus.publish_inbound(_inbound())
-            await _wait_for(lambda: manager._client.runs.stream.call_count == 2)
+            await asyncio.wait_for(bus.join_inbound(), timeout=5)
             await manager.stop()
 
-            assert manager._client.runs.stream.call_count == 2
+            # Before admission the retry runs; after admission it is a duplicate
+            # of a run the Gateway is still executing and must be dropped.
+            assert manager._client.runs.stream.call_count == (1 if admitted else 2)
 
         _run(go())
 
@@ -1947,7 +1959,7 @@ class TestChannelManager:
 
             history_by_checkpoint: dict[tuple[str, str], list[str]] = {}
 
-            async def _runs_wait(thread_id, assistant_id, *, input, config, context, multitask_strategy=None, metadata=None):
+            async def _runs_wait(thread_id, assistant_id, *, input, config, context, multitask_strategy=None, metadata=None, on_disconnect=None):
                 del assistant_id, context  # unused in this test, kept for signature parity
 
                 checkpoint_ns = config.get("configurable", {}).get("checkpoint_ns")
@@ -5548,6 +5560,10 @@ class TestGithubFireAndForget:
             # multitask_strategy must still be ``reject`` — concurrent runs on
             # the same GitHub thread are surfaced via ConflictError below.
             assert create_args[1]["multitask_strategy"] == "reject"
+            # ``runs.create`` returns as soon as the run is pending, so there is
+            # no long-lived request whose disconnect could cancel the run — and
+            # the SDK's ``create`` has no ``on_disconnect`` parameter at all.
+            assert "on_disconnect" not in create_args[1]
 
         _run(go())
 
@@ -5729,6 +5745,249 @@ async def _stale_pin_drain_setup(thread_id, *, model, pin=None, session_model=No
     return manager, carrier_msg, mock_client, bus
 
 
+class TestChannelRunDisconnectPolicy:
+    """IM runs must outlive the HTTP request that created them.
+
+    The manager holds ``runs.wait`` / ``runs.stream`` open for the whole run,
+    and that request dies with the receiving Gateway instance: a leader
+    hand-off, a rolling restart, a crash. The Gateway's default
+    ``on_disconnect`` is ``cancel``, which would cancel the run itself, so
+    every long-lived channel request asks for ``continue``: the run finishes
+    and persists its thread state even though the IM reply is lost with the
+    transport.
+    """
+
+    def test_runs_wait_asks_gateway_to_continue_on_disconnect(self):
+        from app.channels.manager import CHANNEL_RUN_ON_DISCONNECT, ChannelManager
+
+        assert CHANNEL_RUN_ON_DISCONNECT == "continue"
+
+        async def go():
+            bus = MessageBus()
+            store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+            mock_client = _make_mock_langgraph_client(thread_id="slack-thread-1")
+            manager._client = mock_client
+
+            await manager._handle_chat(InboundMessage(channel_name="slack", chat_id="C123", user_id="U1", text="hi"))
+
+            mock_client.runs.wait.assert_called_once()
+            wait_kwargs = mock_client.runs.wait.call_args.kwargs
+            assert wait_kwargs["on_disconnect"] == "continue"
+            assert wait_kwargs["multitask_strategy"] == "reject"
+
+        _run(go())
+
+    def test_runs_stream_asks_gateway_to_continue_on_disconnect(self):
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+            captured: dict[str, object] = {}
+
+            async def _stream(thread_id, assistant_id, **kwargs):  # noqa: ARG001
+                captured.update(kwargs)
+                yield _make_stream_part(
+                    "values",
+                    {
+                        "messages": [
+                            {"type": "human", "content": "hi"},
+                            {"type": "ai", "content": "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._handle_chat(InboundMessage(channel_name="feishu", chat_id="chat1", user_id="user1", text="hi"))
+
+            mock_client.runs.stream.assert_called_once()
+            assert captured["on_disconnect"] == "continue"
+            assert captured["multitask_strategy"] == "reject"
+
+        _run(go())
+
+
+class TestDedupeKeySurvivesAdmittedRun:
+    """A provider redelivery must not start a second run for a turn the
+    Gateway may already be executing.
+
+    Channel runs are created with ``on_disconnect: continue`` (see
+    ``TestChannelRunDisconnectPolicy``), so a run outlives the manager worker
+    that requested it. Releasing the inbound dedupe key when that worker is
+    cancelled (a leader hand-off, a shutdown) would let the provider's
+    redelivery through on the replacement instance and execute the same turn
+    twice. The key is therefore released only while the manager still owns the
+    outcome: before the run request leaves it, or when the Gateway provably
+    rejected the request.
+    """
+
+    @staticmethod
+    def _message(channel: str) -> InboundMessage:
+        if channel == "github":
+            # Shaped like app.gateway.github.dispatcher.fanout_event output.
+            return InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="alice",
+                owner_user_id="alice",
+                text="@bot please review",
+                topic_id="7:reviewer",
+                workspace_id="zhfeng/llm-gateway",
+                metadata={"message_id": "d1:alice:reviewer", "agent_name": "reviewer"},
+            )
+        return InboundMessage(channel_name=channel, chat_id="C1", user_id="U1", text="hi", metadata={"event_id": "evt-1", "team_id": "T1"})
+
+    @staticmethod
+    def _manager(store: MemoryInboundDedupeStore, bus: MessageBus):
+        from app.channels.manager import ChannelManager
+
+        return ChannelManager(
+            bus=bus,
+            store=JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"),
+            inbound_dedupe_store=store,
+            max_concurrency=1,
+            shutdown_grace_period_seconds=0,
+        )
+
+    @staticmethod
+    def _final_stream(*_args, **_kwargs):
+        async def gen():
+            yield _make_stream_part(
+                "values",
+                {"messages": [{"type": "human", "content": "hi"}, {"type": "ai", "content": "done"}], "artifacts": []},
+            )
+
+        return gen()
+
+    async def _hand_off(self, channel: str, *, block: str) -> tuple[MagicMock, bool]:
+        """Cancel instance A while it is blocked at *block*, then redeliver on B.
+
+        Returns B's LangGraph client and whether the dedupe key was still
+        recorded between A's cancellation and B's redelivery.
+        """
+        import app.gateway.github.run_policy  # noqa: F401 — register the fire-and-forget policy
+        from app.channels.manager import ChannelManager
+
+        store = MemoryInboundDedupeStore()
+        blocked = asyncio.Event()
+
+        async def _block(*_args, **_kwargs):
+            blocked.set()
+            await asyncio.Event().wait()
+
+        def _blocking_stream(*_args, **_kwargs):
+            async def gen():
+                yield _make_stream_part(
+                    "messages-tuple",
+                    [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}],
+                )
+                blocked.set()
+                await asyncio.Event().wait()
+
+            return gen()
+
+        bus_a = MessageBus()
+        manager_a = self._manager(store, bus_a)
+        client_a = _make_mock_langgraph_client(thread_id="thread-a")
+        if block == "thread_create":
+            client_a.threads.create = AsyncMock(side_effect=_block)
+        elif block == "wait":
+            client_a.runs.wait = AsyncMock(side_effect=_block)
+        elif block == "stream":
+            client_a.runs.stream = MagicMock(side_effect=_blocking_stream)
+        elif block == "create":
+            client_a.runs.create = AsyncMock(side_effect=_block)
+        else:
+            raise AssertionError(block)
+        manager_a._client = client_a
+        await manager_a.start()
+        await bus_a.publish_inbound(self._message(channel))
+        await asyncio.wait_for(blocked.wait(), timeout=5)
+        # Grace period 0: stop() cancels the worker in the middle of the handler,
+        # exactly what a leader hand-off or a shutdown does.
+        await manager_a.stop()
+
+        key = ChannelManager._inbound_dedupe_key(self._message(channel))
+        assert key is not None
+        recorded_after_cancel = key in store._store
+
+        bus_b = MessageBus()
+        manager_b = self._manager(store, bus_b)
+        client_b = _make_mock_langgraph_client(thread_id="thread-b")
+        client_b.runs.stream = MagicMock(side_effect=self._final_stream)
+        client_b.runs.create = AsyncMock(return_value={"run_id": "run-b", "status": "pending"})
+        manager_b._client = client_b
+        await manager_b.start()
+        await bus_b.publish_inbound(self._message(channel))
+        await asyncio.wait_for(bus_b.join_inbound(), timeout=5)
+        await manager_b.stop()
+        return client_b, recorded_after_cancel
+
+    @staticmethod
+    def _run_requests(client: MagicMock) -> int:
+        return client.runs.wait.call_count + client.runs.stream.call_count + client.runs.create.call_count
+
+    @pytest.mark.parametrize(
+        ("channel", "block"),
+        [("slack", "wait"), ("feishu", "stream"), ("github", "create")],
+        ids=["slack-wait", "feishu-stream", "github-create"],
+    )
+    def test_cancel_after_the_run_request_keeps_the_key_so_redelivery_is_dropped(self, channel, block):
+        async def go():
+            client_b, recorded_after_cancel = await self._hand_off(channel, block=block)
+            assert recorded_after_cancel is True
+            assert self._run_requests(client_b) == 0
+
+        _run(go())
+
+    @pytest.mark.parametrize("channel", ["slack", "feishu", "github"])
+    def test_cancel_before_the_run_request_releases_the_key_so_redelivery_retries(self, channel):
+        async def go():
+            client_b, recorded_after_cancel = await self._hand_off(channel, block="thread_create")
+            assert recorded_after_cancel is False
+            assert self._run_requests(client_b) == 1
+
+        _run(go())
+
+    @pytest.mark.parametrize(
+        ("error", "kept"),
+        [
+            (httpx.ReadTimeout("read timed out"), True),
+            (httpx.RemoteProtocolError("peer closed connection"), True),
+            (httpx.ConnectError("connection refused"), False),
+            (RuntimeError("unexpected"), False),
+        ],
+        ids=["read-timeout-keeps", "protocol-error-keeps", "connect-error-releases", "unexpected-error-releases"],
+    )
+    def test_wait_failure_releases_the_key_only_when_the_gateway_cannot_have_admitted_the_run(self, error, kept):
+        """``/wait`` confirms admission only with its final response, so a
+        failure after the request was read by the Gateway (a read timeout, a
+        dropped connection) must keep the key: the run is still executing."""
+
+        async def go():
+            from app.channels.manager import ChannelManager
+
+            store = MemoryInboundDedupeStore()
+            manager = self._manager(store, MessageBus())
+            client = _make_mock_langgraph_client(thread_id="thread-1")
+            client.runs.wait = AsyncMock(side_effect=error)
+            manager._client = client
+            msg = self._message("slack")
+            assert await manager._is_duplicate_inbound(msg) is False  # recorded on receipt, as the worker does
+
+            await manager._handle_message(msg)
+
+            assert (ChannelManager._inbound_dedupe_key(msg) in store._store) is kept
+
+        _run(go())
+
+
 class TestGithubFollowupBuffer:
     """Tests for issue #4121 Slice 2: buffer-and-drain of concurrent GitHub
     comments that arrive while a run is already active on the thread.

deploy/helm/deer-flow/README.md

@@ -309,7 +309,8 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002
   next heartbeat. The startup gate refuses `enabled: false` while channels are
   enabled. Credentials saved from the web UI reach the holding Pod only when it
   next starts the channel, so restart it after changing them, and a failover
-  cancels the IM replies in flight on the old holder. `gateway.replicas: 1`
+  loses the IM replies in flight on the old holder, though the runs themselves
+  finish and their threads stay readable in the web UI. `gateway.replicas: 1`
   stays the default because the WeChat QR login and browser tools are still
   single-instance. Before raising replicas, also set
   `persistence.home.accessMode: ReadWriteMany` on multi-node clusters (thread

frontend/src/content/en/application/deployment-guide.mdx

@@ -156,7 +156,7 @@ The Helm chart ships `gateway.replicas: 1`. Raising it is supported for the web
 
 The chart exports `DEER_FLOW_MULTI_INSTANCE=true` when `gateway.replicas > 1`, and the Gateway refuses to start if any of these is missing. If you scale with `kubectl scale` or an HPA instead of `gateway.replicas`, set `gateway.multiInstance: true` first so the declaration, the PodDisruptionBudget and the required shared `AUTH_JWT_SECRET` do not keep their single-replica rendering (the PodDisruptionBudget then blocks node drains of the single Pod until the scale-up happens; disable it with `gateway.podDisruptionBudget.enabled: false` for that interim if needed). On multi-node clusters also set `persistence.home.accessMode: ReadWriteMany` (uploads, outputs, memory and `extensions_config.json` live on that volume; MCP and skill changes made through one Pod are then visible to the others on their next request) and `agent_storage.backend: db` so custom agents are visible on every Pod.
 
-IM channels work on several replicas through `channels.leader_lease` in `config` (default `auto`): one Pod holds each platform connection, and when it dies another Pod takes over within `lease_seconds + grace_seconds` plus one renewal interval (50 seconds by default). The startup gate refuses `enabled: false` while channels are enabled. Channel credentials saved from the web UI reach the Pod holding that platform only when it next starts the channel, so restart that Pod after changing them; a takeover cancels the IM replies the old Pod was still generating. Still single-instance: the WeChat QR login and browser tools. Keep one replica while those are enabled.
+IM channels work on several replicas through `channels.leader_lease` in `config` (default `auto`): one Pod holds each platform connection, and when it dies another Pod takes over within `lease_seconds + grace_seconds` plus one renewal interval (50 seconds by default). The startup gate refuses `enabled: false` while channels are enabled. Channel credentials saved from the web UI reach the Pod holding that platform only when it next starts the channel, so restart that Pod after changing them; a takeover loses the IM replies the old Pod was still delivering, although the runs themselves finish and their threads stay readable in the web UI. Still single-instance: the WeChat QR login and browser tools. Keep one replica while those are enabled.
 
 ### K8s Provisioner setup
 

frontend/src/content/zh/application/deployment-guide.mdx

@@ -153,7 +153,7 @@ Helm chart 默认 `gateway.replicas: 1`。当所有副本共享同一个 Postgre
 
 当 `gateway.replicas > 1` 时，chart 会导出 `DEER_FLOW_MULTI_INSTANCE=true`，缺少上述任一项时 Gateway 会拒绝启动。如果你通过 `kubectl scale` 或 HPA 而不是 `gateway.replicas` 扩容，请先设置 `gateway.multiInstance: true`：它们在不执行 Helm upgrade 的情况下改变副本数，否则该声明、PodDisruptionBudget 和必需的共享 `AUTH_JWT_SECRET` 都会停留在单副本的渲染结果上（在真正扩容之前，PodDisruptionBudget 会阻止驱逐唯一的 Pod 和排空节点；如有需要，可在此期间设置 `gateway.podDisruptionBudget.enabled: false`）。多节点集群还需要把 `persistence.home.accessMode` 设为 `ReadWriteMany`（上传、产物、记忆与 `extensions_config.json` 都在该卷上；在一个 Pod 上修改的 MCP 与技能配置会在其他 Pod 的下一次请求中生效），并设置 `agent_storage.backend: db` 让自定义 agent 在每个 Pod 上可见。
 
-IM 通道可以通过 `config` 中的 `channels.leader_lease`（默认 `auto`）在多副本下运行：每个平台的连接只由一个 Pod 持有，该 Pod 宕机后，另一个 Pod 会在 `lease_seconds + grace_seconds` 再加一个续约周期内接管（默认 50 秒）。启用通道时启动门禁会拒绝 `enabled: false`。在 Web UI 中保存的通道凭据要等持有该平台的 Pod 下次启动通道时才会生效，修改后请重启该 Pod；接管会取消旧 Pod 上仍在生成的 IM 回复。仍然只能单实例运行的部分：WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。
+IM 通道可以通过 `config` 中的 `channels.leader_lease`（默认 `auto`）在多副本下运行：每个平台的连接只由一个 Pod 持有，该 Pod 宕机后，另一个 Pod 会在 `lease_seconds + grace_seconds` 再加一个续约周期内接管（默认 50 秒）。启用通道时启动门禁会拒绝 `enabled: false`。在 Web UI 中保存的通道凭据要等持有该平台的 Pod 下次启动通道时才会生效，修改后请重启该 Pod；接管会丢失旧 Pod 上仍在投递的 IM 回复，但运行本身会继续完成，其对话线程仍可在 Web UI 中查看。仍然只能单实例运行的部分：WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。
 
 ### K8s Provisioner 配置