fix(channels): keep config reload publication under lifecycle lock (#6390)

* fix(channels): keep config reload publication under lifecycle lock

* test(channels): prevent stale config publish after cancellation

* test(channels): exercise real stale reload write

* test(channels): harden cancelled config reload regression

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -780,6 +780,8 @@ cleanup when migrating from legacy metadata credentials.
 
 DeerFlow supports receiving tasks from messaging apps. Channels auto-start when configured — no public IP required for any of them.
 
+Cancelling a channel restart discards its pending configuration reload, preserving newer runtime settings applied afterward.
+
 DeerFlow can also expose user-owned IM channel connections in the workspace UI. When `channel_connections` is enabled, logged-in users can bind Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, WeCom, QQ, or Buzz from the sidebar / Settings > Channels. It reuses the existing outbound `channels.*` transports, so no public IP or provider callback URL is required. Incoming IM messages then run under the connected DeerFlow user account. See [IM Channel Connections](backend/docs/IM_CHANNEL_CONNECTIONS.md) for setup and security notes.
 
 | Channel | Transport | Difficulty |

backend/app/channels/service.py

@@ -334,7 +334,8 @@ class ChannelService:
         The UI runtime-config overlay applied at startup is re-applied here
         so a file-driven reload neither drops credentials entered from the
         browser nor resurrects a channel disconnected from it.
-        Falls back to the cached ``self._config`` when config loading fails.
+        Returns a snapshot without mutating shared state, or ``None`` when
+        unavailable. The restart caller owns publication and cached fallback.
         """
         try:
             from deerflow.config.app_config import get_app_config
@@ -345,12 +346,10 @@ class ChannelService:
             _merge_channel_connection_runtime_config(channels_config, app_config)
             channel_config = channels_config.get(name)
             if isinstance(channel_config, dict):
-                # Update the cached config so get_status() stays consistent.
-                self._config[name] = channel_config
                 return channel_config
         except Exception:
-            logger.exception("Failed to reload config for channel %s, using cached version", name)
-        return self._config.get(name)
+            logger.exception("Failed to load config snapshot for channel %s", name)
+        return None
 
     def _channel_lock(self, name: str) -> asyncio.Lock:
         """Return the per-channel lifecycle lock ensuring serialized mutations."""
@@ -384,7 +383,11 @@ class ChannelService:
         if reload_config:
             # Reading config.yaml and the runtime store is disk IO; keep it
             # off the event loop.
-            config = await asyncio.to_thread(self._load_channel_config, name)
+            loaded_config = await asyncio.to_thread(self._load_channel_config, name)
+            # Publish under the lifecycle lock after await; abandoned workers stay read-only.
+            if loaded_config is not None:
+                self._config[name] = loaded_config
+            config = self._config.get(name)
             self._bump_config_epoch(name)
         else:
             config = self._config.get(name)

backend/tests/AGENTS.md

@@ -2,6 +2,10 @@
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+Channel reload cancellation regressions must assert the worker returned the
+stale snapshot before checking that newer runtime config survived. Completion
+alone cannot prove the race: loader exceptions are caught and return `None`.
+
 Browser egress session-close tests retain the real listener and its sockets,
 verify a SOCKS handshake, then check listener shutdown, closed socket descriptors
 and EOF on the established client. A new connection to the old port is not a

backend/tests/test_channel_config_reload_cancellation.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from app.channels.service import ChannelService
+from deerflow.config import app_config as app_config_module
+
+
+@pytest.mark.asyncio
+async def test_cancelled_reload_cannot_overwrite_new_runtime_config(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = ChannelService(
+        channels_config={
+            "wechat": {
+                "enabled": False,
+                "marker": "initial",
+            }
+        }
+    )
+    reload_started = threading.Event()
+    allow_reload = threading.Event()
+    reload_finished = threading.Event()
+    loaded = []
+
+    def blocking_get_app_config():
+        reload_started.set()
+        assert allow_reload.wait(timeout=2)
+        return SimpleNamespace(
+            model_extra={
+                "channels": {
+                    "wechat": {
+                        "enabled": False,
+                        "marker": "stale",
+                    }
+                }
+            },
+            channel_connections=None,
+        )
+
+    monkeypatch.setattr(app_config_module, "get_app_config", blocking_get_app_config)
+
+    real_load = service._load_channel_config
+
+    def tracked_load(name: str):
+        try:
+            config = real_load(name)
+            loaded.append(config)
+            return config
+        finally:
+            reload_finished.set()
+
+    monkeypatch.setattr(service, "_load_channel_config", tracked_load)
+
+    restart = asyncio.create_task(service.restart_channel("wechat"))
+    try:
+        assert await asyncio.to_thread(reload_started.wait, 2)
+        restart.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await restart
+
+        assert await service.configure_channel(
+            "wechat",
+            {"enabled": False, "marker": "fresh"},
+        )
+
+        allow_reload.set()
+        assert await asyncio.to_thread(reload_finished.wait, 2)
+        assert loaded == [{"enabled": False, "marker": "stale"}], "worker did not produce the stale config snapshot"
+        assert service._config["wechat"]["marker"] == "fresh"
+    finally:
+        allow_reload.set()
+        if not reload_finished.is_set():
+            await asyncio.to_thread(reload_finished.wait, 2)
+        if not restart.done():
+            restart.cancel()
+            await asyncio.gather(restart, return_exceptions=True)