diff --git a/backend/app/gateway/routers/channel_connections.py b/backend/app/gateway/routers/channel_connections.py index 1c7133078..ea57f5126 100644 --- a/backend/app/gateway/routers/channel_connections.py +++ b/backend/app/gateway/routers/channel_connections.py @@ -412,7 +412,16 @@ def _provider_response( from app.gateway.auth_disabled import is_auth_disabled status, unavailable_reason = _provider_status(config, channels_config, provider) - if connection: + if unavailable_reason is not None: + # The runtime provider is unavailable, so a stale "connected" row must + # not be reported as connected. Other statuses (e.g. "revoked") are + # preserved so consumers can still distinguish a revoked binding from a + # never-connected one. + if connection and connection["status"] != "connected": + connection_status = connection["status"] + else: + connection_status = "not_connected" + elif connection: connection_status = connection["status"] elif is_auth_disabled() and status["configured"] and unavailable_reason is None: # Auth-disabled local mode routes every channel message to the default @@ -561,7 +570,6 @@ async def disconnect_channel_provider_runtime(provider: str, request: Request) - if not provider_config.enabled: raise HTTPException(status_code=400, detail="Channel provider is not enabled") - owner_user_id = _get_user_id(request) try: repo = _get_repository(request, config) except HTTPException as exc: @@ -569,25 +577,33 @@ async def disconnect_channel_provider_runtime(provider: str, request: Request) - raise repo = None - if repo is not None: - for connection in await repo.list_connections(owner_user_id): - if connection["provider"] == provider and connection["status"] != "revoked": - await repo.disconnect_connection( - connection_id=connection["id"], - owner_user_id=owner_user_id, - ) + current_channels_config = await _get_channels_config(request) + candidate_channels_config = dict(current_channels_config) + candidate_channels_config.pop(provider, None) - store = await _get_runtime_config_store(request) - await asyncio.to_thread(store.set_provider_disconnected, provider) - channels_config = await _load_channels_config(request, config) - request.app.state.channels_config = channels_config - - stopped = await _sync_runtime_channel_after_removal(provider, channels_config) + stopped = await _sync_runtime_channel_after_removal(provider, candidate_channels_config) if stopped is False: display_name = _PROVIDER_META[provider]["display_name"] raise HTTPException(status_code=400, detail=f"Failed to stop {display_name} channel. Try again.") - return _provider_response(config, channels_config, provider, _PROVIDER_META[provider]) + # Revoke the DB connection rows before committing the store/cache so a repo + # failure cannot leave the store and cache saying "disconnected" while the + # DB still holds "connected" rows that a later re-configure would silently + # reactivate. + 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 + + return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider]) @router.post("/{provider}/connect", response_model=ChannelConnectResponse) @@ -656,8 +672,8 @@ async def configure_channel_provider_runtime( # cached by get_app_config(). runtime_config["bot_username"] = values["bot_username"] - channels_config[provider] = runtime_config - request.app.state.channels_config = channels_config + candidate_channels_config = dict(channels_config) + candidate_channels_config[provider] = runtime_config started = await _restart_runtime_channel_if_available(provider, runtime_config) if started is False: @@ -667,4 +683,11 @@ async def configure_channel_provider_runtime( store = await _get_runtime_config_store(request) await asyncio.to_thread(store.set_provider_config, provider, runtime_config) - return _provider_response(config, channels_config, provider, _PROVIDER_META[provider]) + # 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 + + return _provider_response(config, live_channels_config, provider, _PROVIDER_META[provider]) diff --git a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py b/backend/packages/harness/deerflow/persistence/channel_connections/sql.py index 4739fd3c9..bc71926fa 100644 --- a/backend/packages/harness/deerflow/persistence/channel_connections/sql.py +++ b/backend/packages/harness/deerflow/persistence/channel_connections/sql.py @@ -177,6 +177,24 @@ class ChannelConnectionRepository: await session.commit() return True + async def disconnect_provider_connections(self, *, provider: str) -> int: + """Revoke all active user connections for an instance-wide provider removal.""" + async with self.session_factory() as session: + result = await session.execute( + select(ChannelConnectionRow.id).where( + ChannelConnectionRow.provider == provider, + ChannelConnectionRow.status != "revoked", + ) + ) + connection_ids = [row_id for row_id in result.scalars()] + if not connection_ids: + return 0 + + await session.execute(update(ChannelConnectionRow).where(ChannelConnectionRow.id.in_(connection_ids)).values(status="revoked")) + await session.execute(delete(ChannelCredentialRow).where(ChannelCredentialRow.connection_id.in_(connection_ids))) + await session.commit() + return len(connection_ids) + async def store_credentials( self, connection_id: str, diff --git a/backend/tests/test_channel_connections_router.py b/backend/tests/test_channel_connections_router.py index f4915fac8..638fbf215 100644 --- a/backend/tests/test_channel_connections_router.py +++ b/backend/tests/test_channel_connections_router.py @@ -294,6 +294,41 @@ def test_get_providers_reports_configured_channel_not_running(tmp_path, monkeypa anyio.run(repo.close) +def test_get_providers_provider_unavailable_overrides_stale_connected_row(tmp_path): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + + async def seed_connection(): + await repo.upsert_connection( + owner_user_id=str(_user().id), + provider="slack", + external_account_id="U123", + status="connected", + ) + + anyio.run(seed_connection) + config = ChannelConnectionsConfig.model_validate( + { + "enabled": True, + "slack": {"enabled": True}, + } + ) + app = _make_app(config, repo, {}) + + with TestClient(app) as client: + response = client.get("/api/channels/providers") + + assert response.status_code == 200 + by_provider = {item["provider"]: item for item in response.json()["providers"]} + assert by_provider["slack"]["configured"] is False + assert by_provider["slack"]["connectable"] is False + assert by_provider["slack"]["connection_status"] == "not_connected" + assert "Slack credentials" in by_provider["slack"]["unavailable_reason"] + + anyio.run(repo.close) + + def test_get_providers_restarts_configured_channel_when_service_can_reconcile(tmp_path, monkeypatch): import anyio @@ -614,6 +649,54 @@ def test_runtime_config_endpoints_require_admin(tmp_path): anyio.run(repo.close) +def test_configure_provider_runtime_rolls_back_visible_state_when_start_fails(tmp_path, monkeypatch): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + config = ChannelConnectionsConfig.model_validate( + { + "enabled": True, + "slack": {"enabled": True}, + } + ) + existing_runtime_config = { + "enabled": True, + "bot_token": "xoxb-old", + "app_token": "xapp-old", + } + runtime_config_store = ChannelRuntimeConfigStore(tmp_path / "channels" / "runtime-config.json") + runtime_config_store.set_provider_config("slack", existing_runtime_config) + service = SimpleNamespace(configure_channel=AsyncMock(return_value=False)) + monkeypatch.setattr("app.channels.service.get_channel_service", lambda: service) + app = _make_app( + config, + repo, + {"slack": dict(existing_runtime_config)}, + runtime_config_store=runtime_config_store, + ) + + with TestClient(app) as client: + response = client.post( + "/api/channels/slack/runtime-config", + json={"values": {"bot_token": "xoxb-new", "app_token": "xapp-new"}}, + ) + + assert response.status_code == 400 + assert "Failed to start Slack channel" in response.json()["detail"] + service.configure_channel.assert_awaited_once_with( + "slack", + { + "enabled": True, + "bot_token": "xoxb-new", + "app_token": "xapp-new", + }, + ) + assert app.state.channels_config["slack"] == existing_runtime_config + assert runtime_config_store.get_provider_config("slack") == existing_runtime_config + + anyio.run(repo.close) + + def test_configure_telegram_runtime_uses_new_bot_username_for_deep_link_without_mutating_config(tmp_path): import anyio @@ -862,7 +945,7 @@ def test_disconnect_provider_runtime_config_suppresses_file_config_and_stops_cha anyio.run(repo.close) -def test_disconnect_provider_runtime_config_revokes_current_user_provider_connections(tmp_path): +def test_disconnect_provider_runtime_config_revokes_all_provider_connections(tmp_path): import anyio repo = anyio.run(_make_repo, tmp_path) @@ -874,6 +957,18 @@ def test_disconnect_provider_runtime_config_revokes_current_user_provider_connec external_account_id="U123", status="connected", ) + await repo.upsert_connection( + owner_user_id="other-user", + provider="slack", + external_account_id="U456", + status="connected", + ) + await repo.upsert_connection( + owner_user_id="other-user", + provider="telegram", + external_account_id="42", + status="connected", + ) anyio.run(seed_connection) config = ChannelConnectionsConfig.model_validate( @@ -895,10 +990,132 @@ def test_disconnect_provider_runtime_config_revokes_current_user_provider_connec assert configure_response.status_code == 200 assert disconnect_response.status_code == 200 - async def get_connection_status(): - return (await repo.list_connections(str(_user().id)))[0]["status"] + async def get_connection_statuses(): + return { + "admin_slack": (await repo.list_connections(str(_user().id)))[0]["status"], + "other": {item["provider"]: item["status"] for item in await repo.list_connections("other-user")}, + } - assert anyio.run(get_connection_status) == "revoked" + statuses = anyio.run(get_connection_statuses) + assert statuses["admin_slack"] == "revoked" + assert statuses["other"]["slack"] == "revoked" + assert statuses["other"]["telegram"] == "connected" + + anyio.run(repo.close) + + +def test_get_providers_preserves_revoked_status_when_provider_unavailable(tmp_path): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + + async def seed_connection(): + await repo.upsert_connection( + owner_user_id=str(_user().id), + provider="slack", + external_account_id="U123", + status="revoked", + ) + + anyio.run(seed_connection) + config = ChannelConnectionsConfig.model_validate( + { + "enabled": True, + "slack": {"enabled": True}, + } + ) + # No runtime channels_config -> the slack provider is unavailable. + app = _make_app(config, repo, {}) + + with TestClient(app) as client: + response = client.get("/api/channels/providers") + + assert response.status_code == 200 + by_provider = {item["provider"]: item for item in response.json()["providers"]} + assert by_provider["slack"]["connectable"] is False + assert by_provider["slack"]["unavailable_reason"] is not None + # A revoked binding must stay distinguishable from a never-connected one, + # even when the runtime provider is currently unavailable. + assert by_provider["slack"]["connection_status"] == "revoked" + + anyio.run(repo.close) + + +def test_configure_provider_runtime_does_not_clobber_concurrent_config_update(tmp_path, monkeypatch): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + config = ChannelConnectionsConfig.model_validate( + { + "enabled": True, + "slack": {"enabled": True}, + "telegram": {"enabled": True, "bot_username": "deerflow_bot"}, + } + ) + runtime_config_store = ChannelRuntimeConfigStore(tmp_path / "channels" / "runtime-config.json") + app = _make_app(config, repo, {}, runtime_config_store=runtime_config_store) + + async def configure_channel(provider, runtime_config): + # Simulate a concurrent admin request for a *different* provider whose + # write to app.state lands while this request awaits the worker restart. + app.state.channels_config = { + **app.state.channels_config, + "telegram": {"enabled": True, "bot_token": "tg-token"}, + } + return True + + service = SimpleNamespace(configure_channel=configure_channel) + monkeypatch.setattr("app.channels.service.get_channel_service", lambda: service) + + with TestClient(app) as client: + response = client.post( + "/api/channels/slack/runtime-config", + json={"values": {"bot_token": "xoxb-ui", "app_token": "xapp-ui"}}, + ) + + assert response.status_code == 200 + # The concurrent telegram write must survive alongside the slack write. + assert app.state.channels_config["slack"]["bot_token"] == "xoxb-ui" + assert app.state.channels_config["telegram"]["bot_token"] == "tg-token" + + anyio.run(repo.close) + + +def test_disconnect_provider_runtime_keeps_state_consistent_when_revoke_fails(tmp_path): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + config = ChannelConnectionsConfig.model_validate( + { + "enabled": True, + "slack": {"enabled": True}, + } + ) + runtime_config_store = ChannelRuntimeConfigStore(tmp_path / "channels" / "runtime-config.json") + app = _make_app(config, repo, {}, runtime_config_store=runtime_config_store) + + with TestClient(app) as client: + configure_response = client.post( + "/api/channels/slack/runtime-config", + json={"values": {"bot_token": "xoxb-ui", "app_token": "xapp-ui"}}, + ) + assert configure_response.status_code == 200 + + repo.disconnect_provider_connections = AsyncMock(side_effect=RuntimeError("db down")) + + with TestClient(app, raise_server_exceptions=False) as client: + disconnect_response = client.delete("/api/channels/slack/runtime-config") + + assert disconnect_response.status_code == 500 + # When the DB revoke fails, the store/cache must not be left diverged from + # the DB: the provider stays configured so a later re-configure cannot + # silently reactivate un-revoked connection rows. + assert app.state.channels_config["slack"]["bot_token"] == "xoxb-ui" + assert runtime_config_store.get_provider_config("slack") == { + "enabled": True, + "bot_token": "xoxb-ui", + "app_token": "xapp-ui", + } anyio.run(repo.close) diff --git a/frontend/src/components/workspace/channels/workspace-channels-list.tsx b/frontend/src/components/workspace/channels/workspace-channels-list.tsx index 9a0429373..9e361f945 100644 --- a/frontend/src/components/workspace/channels/workspace-channels-list.tsx +++ b/frontend/src/components/workspace/channels/workspace-channels-list.tsx @@ -117,7 +117,9 @@ export function WorkspaceChannelsList() { {visibleProviders.map((provider) => { const canEditRuntimeConfig = providerCanEditRuntimeConfig(provider); - const isConnected = provider.connection_status === "connected"; + const isConnected = + !provider.unavailable_reason && + provider.connection_status === "connected"; const isPending = (connectMutation.isPending && connectMutation.variables === provider.provider) || diff --git a/frontend/src/components/workspace/settings/channels-settings-page.tsx b/frontend/src/components/workspace/settings/channels-settings-page.tsx index e554348d3..c3c830285 100644 --- a/frontend/src/components/workspace/settings/channels-settings-page.tsx +++ b/frontend/src/components/workspace/settings/channels-settings-page.tsx @@ -117,9 +117,11 @@ function ChannelProviderItem({ const configureMutation = useConfigureChannelProvider(); const disconnectProviderMutation = useDisconnectChannelProvider(); const [setupOpen, setSetupOpen] = useState(false); + const runtimeAvailable = provider.configured && !provider.unavailable_reason; const isConnected = - connection?.status === "connected" || - provider.connection_status === "connected"; + runtimeAvailable && + (connection?.status === "connected" || + provider.connection_status === "connected"); const canEditRuntimeConfig = providerCanEditRuntimeConfig(provider); const canConnect = (provider.connectable ?? (provider.enabled && provider.configured)) && @@ -186,7 +188,7 @@ function ChannelProviderItem({ {getProviderDescription(provider, t.channels.descriptions)} - {connectionLabel + {isConnected && connectionLabel ? ` ${t.channels.connectedAs(connectionLabel)}` : ""} {!isConnected && provider.unavailable_reason