fix(channels): keep runtime config commit owned across cancellation (#6198)

* fix(channels): own runtime config commit across cancellation

* test(channels): pin cancelled runtime config commit

* docs(channels): record runtime config commit ownership

* style(tests): fix channel connection router lint

---------

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

backend/app/channels/AGENTS.md

@@ -63,6 +63,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
 - WeCom outbound content is capped at the protocol's 20480 UTF-8 bytes: stream replies clip on a character boundary with a truncation marker (one stream carries the whole reply and cannot split mid-way), while proactive pushes split into at most 10 sequential markdown messages per push, with the remaining tail clipped and marker-terminated. Both paths measure bytes, not characters. A per-chat send lock serializes each split batch end to end, because manager workers run concurrently and two long pushes to the same chat would otherwise interleave chunks; locks are reference-counted and reclaimed once no sender holds or waits on them, so the registry does not grow for the life of the Gateway.
 - Frontend APIs: `GET /api/channels/providers`, `GET /api/channels/connections`, `POST /api/channels/{provider}/connect`, and `DELETE /api/channels/connections/{connection_id}`.
 - Browser APIs remain protected by normal Gateway auth/CSRF. Provider messages arrive through the already-configured channel workers.
+- Runtime credential configure/disconnect treats the runtime-config file write and `app.state.channels_config` reconciliation as one cancellation-owned commit; do not let caller cancellation escape between them.
 - Provider-level `connection_status` reflects the user's newest connection row. With no binding it is `not_connected`, except in auth-disabled local mode where a configured running channel reports `connected` because all channel messages already route to the default user.
 - Slack replies use the configured operator bot token from `channels.slack` unless per-connection credentials are present; unreadable or corrupt stored credentials are treated as unavailable.
 - Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom workers resolve incoming platform identities to connection records before reaching `ChannelManager`.

backend/app/gateway/routers/channel_connections.py

@@ -19,9 +19,11 @@ from app.channels.runtime_config_store import (
 )
 from app.channels.wechat_qr_login import QRLoginError, WechatQRLogin
 from app.gateway.deps import require_admin_user
+from app.gateway.persistent_writes import run_drained_write
 from deerflow.config.channel_connections_config import ChannelConnectionsConfig
 from deerflow.persistence.channel_connections import ChannelConnectionRepository
 from deerflow.persistence.engine import get_session_factory
+from deerflow.utils.file_io import await_drained
 
 router = APIRouter(prefix="/api/channels", tags=["channel-connections"])
 logger = logging.getLogger(__name__)
@@ -623,15 +625,7 @@ async def _disconnect_channel_provider_runtime(provider: str, request: Request)
     if repo is not None:
         await repo.disconnect_provider_connections(provider=provider)
 
-    store = await _get_runtime_config_store(request)
-    await asyncio.to_thread(store.set_provider_disconnected, provider)
-
-    # Re-read the live cached config and drop only this provider so a concurrent
-    # mutation for a different provider is not clobbered. No await may occur
-    # between this read and the reassignment.
-    live_channels_config = await _get_channels_config(request)
-    live_channels_config.pop(provider, None)
-    request.app.state.channels_config = live_channels_config
+    live_channels_config = await _commit_runtime_channel_config(request, provider, None)
 
     return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider])
 
@@ -712,21 +706,41 @@ async def _configure_channel_provider_runtime(provider: str, body: ChannelRuntim
     return await _apply_runtime_channel_config(request, config, provider, runtime_config)
 
 
+async def _commit_runtime_channel_config(
+    request: Request,
+    provider: str,
+    runtime_config: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Persist one runtime config change and reconcile the live cache atomically to cancellation."""
+
+    async def _commit() -> dict[str, Any]:
+        store = await _get_runtime_config_store(request)
+        if runtime_config is None:
+            await run_drained_write("Disconnect channel runtime config", store.set_provider_disconnected, (), provider)
+        else:
+            await run_drained_write("Save channel runtime config", store.set_provider_config, (), provider, runtime_config)
+
+        # Re-read the live cached config after the durable write so concurrent
+        # mutations for sibling providers are preserved. The outer drain keeps
+        # cancellation owned until this in-memory reconciliation also lands.
+        live_channels_config = await _get_channels_config(request)
+        if runtime_config is None:
+            live_channels_config.pop(provider, None)
+        else:
+            live_channels_config[provider] = runtime_config
+        request.app.state.channels_config = live_channels_config
+        return live_channels_config
+
+    return await await_drained(_commit())
+
+
 async def _apply_runtime_channel_config(request: Request, config: ChannelConnectionsConfig, provider: str, runtime_config: dict[str, Any]) -> ChannelProviderResponse:
     started = await _restart_runtime_channel_if_available(provider, runtime_config)
     if started is False:
         display_name = _PROVIDER_META[provider]["display_name"]
         raise HTTPException(status_code=400, detail=f"Failed to start {display_name} channel. Check the values and try again.")
 
-    store = await _get_runtime_config_store(request)
-    await asyncio.to_thread(store.set_provider_config, provider, runtime_config)
-
-    # Re-read the live cached config and apply only this provider's change so a
-    # concurrent mutation for a different provider is not clobbered. No await
-    # may occur between this read and the reassignment.
-    live_channels_config = await _get_channels_config(request)
-    live_channels_config[provider] = runtime_config
-    request.app.state.channels_config = live_channels_config
+    live_channels_config = await _commit_runtime_channel_config(request, provider, runtime_config)
 
     return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider])
 

backend/tests/test_channel_connections_router.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import threading
 from tempfile import TemporaryDirectory
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
@@ -878,6 +880,81 @@ def test_configure_provider_runtime_credentials_preserves_masked_secrets(tmp_pat
     anyio.run(repo.close)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("runtime_config", "expected_persisted"),
+    [
+        ({"enabled": True, "bot_token": "new-token"}, {"enabled": True, "bot_token": "new-token"}),
+        (None, {"enabled": False, "_runtime_disabled": True}),
+    ],
+)
+async def test_runtime_config_commit_drains_persistence_and_cache_across_cancellation(
+    monkeypatch,
+    runtime_config,
+    expected_persisted,
+):
+    started = threading.Event()
+    release = threading.Event()
+    persisted: dict[str, dict] = {}
+    live_channels_config = {
+        "telegram": {"enabled": True, "bot_token": "sibling"},
+        "slack": {"enabled": True, "bot_token": "old-token"},
+    }
+
+    def set_provider_config(provider, value):
+        started.set()
+        assert release.wait(timeout=5)
+        persisted[provider] = dict(value)
+
+    def set_provider_disconnected(provider):
+        started.set()
+        assert release.wait(timeout=5)
+        persisted[provider] = {"enabled": False, "_runtime_disabled": True}
+
+    store = SimpleNamespace(
+        set_provider_config=set_provider_config,
+        set_provider_disconnected=set_provider_disconnected,
+    )
+
+    async def get_store(_request):
+        return store
+
+    async def get_channels_config(_request):
+        return live_channels_config
+
+    monkeypatch.setattr(channel_connections, "_get_runtime_config_store", get_store)
+    monkeypatch.setattr(channel_connections, "_get_channels_config", get_channels_config)
+    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(channels_config=live_channels_config)))
+
+    task = asyncio.create_task(channel_connections._commit_runtime_channel_config(request, "slack", runtime_config))
+    try:
+        assert await asyncio.to_thread(started.wait, 5)
+        task.cancel()
+        await asyncio.sleep(0)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not task.done()
+
+        release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    finally:
+        release.set()
+        if not task.done():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+    assert persisted["slack"] == expected_persisted
+    assert request.app.state.channels_config["telegram"] == {
+        "enabled": True,
+        "bot_token": "sibling",
+    }
+    if runtime_config is None:
+        assert "slack" not in request.app.state.channels_config
+    else:
+        assert request.app.state.channels_config["slack"] == runtime_config
+
+
 def test_disconnect_provider_runtime_config_clears_connected_state(tmp_path):
     import anyio