fix(gateway): Phase 3 — evict pool instance on ServerConfigUpdated

Autonomous decisions:
- UserSpaceSyncService uses optional EventSender via with_event_sender() — keeps existing new() call sites compiling; file-watcher now wired to the app-wide EventBus sender created alongside ServerAppService
- Split regression coverage: user_space_sync event emission test + pool remove_instance unit test — full handler integration test skipped due to heavy mock surface in gateway lib tests
- Added cfg(test) insert_test_instance on PoolService — minimal test seam to assert eviction clears a pooled entry without standing up a live MCP connection
- Amended (not separate commit) — branch is ahead of origin by one unpushed Phase 3 commit; lib.rs wiring is the same logical change

Signed-off-by: crimsonsunset <jsangio1@gmail.com>

Author: crimsonsunset

apps/desktop/src-tauri/src/lib.rs

@@ -643,6 +643,7 @@ pub fn run() {
                 let app_state: tauri::State<'_, AppState> = app.state();
                 let spaces_dir = app_state.spaces_dir().to_path_buf();
                 let installed_repo = app_state.installed_server_repository.clone();
+                let event_sender_for_sync = event_bus.sender();
                 let app_handle_for_watcher = app.handle().clone();
 
                 // Use the well-known default space UUID
@@ -656,7 +657,10 @@ pub fn run() {
                     // Create file watcher with UI event emitter
                     match services::SpaceFileWatcher::new(
                         spaces_dir.clone(),
-                        Arc::new(mcpmux_core::application::UserSpaceSyncService::new(installed_repo)),
+                        Arc::new(
+                            mcpmux_core::application::UserSpaceSyncService::new(installed_repo)
+                                .with_event_sender(event_sender_for_sync),
+                        ),
                         default_space_id,
                         services::SpaceFileWatcherEmitters {
                             on_success: Some(Arc::new(

crates/mcpmux-core/src/application/user_space_sync.rs

@@ -9,9 +9,11 @@ use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use tracing::{debug, info};
+use uuid::Uuid;
 
 use crate::domain::config::UserSpaceConfig;
-use crate::domain::{InstallationSource, InstalledServer, ServerDefinition};
+use crate::domain::{DomainEvent, InstallationSource, InstalledServer, ServerDefinition};
+use crate::event_bus::EventSender;
 use crate::repository::InstalledServerRepository;
 
 /// Result of a sync operation
@@ -54,12 +56,40 @@ enum SyncOutcome {
 /// Service for syncing user space JSON config files to InstalledServer records
 pub struct UserSpaceSyncService {
     installed_repo: Arc<dyn InstalledServerRepository>,
+    event_sender: Option<EventSender>,
 }
 
 impl UserSpaceSyncService {
     /// Create a new sync service
     pub fn new(installed_repo: Arc<dyn InstalledServerRepository>) -> Self {
-        Self { installed_repo }
+        Self {
+            installed_repo,
+            event_sender: None,
+        }
+    }
+
+    /// Attach an event sender so definition updates emit `ServerConfigUpdated`.
+    pub fn with_event_sender(mut self, sender: EventSender) -> Self {
+        self.event_sender = Some(sender);
+        self
+    }
+
+    /// Emit `ServerConfigUpdated` when a wired event sender is present.
+    fn emit_server_config_updated(&self, space_id: &str, server_id: &str) {
+        let Some(sender) = &self.event_sender else {
+            return;
+        };
+        let Ok(space_uuid) = Uuid::parse_str(space_id) else {
+            debug!(
+                "Skipping ServerConfigUpdated emit for {space_id}/{server_id}: invalid space UUID"
+            );
+            return;
+        };
+
+        sender.emit(DomainEvent::ServerConfigUpdated {
+            space_id: space_uuid,
+            server_id: server_id.to_string(),
+        });
     }
 
     /// Ensure no two user-config entries normalize to the same MCP server id.
@@ -197,7 +227,8 @@ impl UserSpaceSyncService {
                 }
                 Ok(SyncOutcome::Updated) => {
                     debug!("Updated server: {}", server_id);
-                    result.updated.push(server_id);
+                    result.updated.push(server_id.clone());
+                    self.emit_server_config_updated(space_id, &server_id);
                 }
                 Ok(SyncOutcome::Adopted) => {
                     info!("Adopted server from another source: {}", server_id);
@@ -687,4 +718,49 @@ mod tests {
             "unexpected error: {err}"
         );
     }
+
+    #[tokio::test]
+    async fn sync_from_file_emits_server_config_updated_on_definition_update() {
+        use crate::EventBus;
+
+        let space_id = "00000000-0000-0000-0000-000000000001";
+        let bus = EventBus::new();
+        let mut receiver = bus.subscribe();
+        let repo = Arc::new(InMemoryInstalledServerRepository::new());
+        let (_dir, file_path) = write_config_file(
+            r#"{ "mcpServers": {
+                "alpha": { "command": "echo", "name": "Alpha v1" }
+            } }"#,
+        )
+        .await;
+
+        repo.seed(
+            InstalledServer::new(space_id, "alpha")
+                .with_source(InstallationSource::UserConfig {
+                    file_path: file_path.clone(),
+                })
+                .with_enabled(true),
+        )
+        .await;
+
+        let service = UserSpaceSyncService::new(repo).with_event_sender(bus.sender());
+        tokio::fs::write(
+            &file_path,
+            r#"{ "mcpServers": {
+                "alpha": { "command": "echo", "name": "Alpha v2" }
+            } }"#,
+        )
+        .await
+        .expect("rewrite config file");
+
+        let result = service
+            .sync_from_file(space_id, &file_path)
+            .await
+            .expect("sync should update existing server");
+        assert_eq!(result.updated, vec!["alpha".to_string()]);
+
+        let event = receiver.recv().await.expect("ServerConfigUpdated should emit");
+        assert_eq!(event.type_name(), "server_config_updated");
+        assert_eq!(event.server_id(), Some("alpha"));
+    }
 }

crates/mcpmux-gateway/src/admin/write_runtime.rs

@@ -178,8 +178,64 @@ impl GatewayWriteRuntime for LiveGatewayWriteRuntime {
         Err(gateway_write_unavailable())
     }
 
-    async fn retry_connection(&self, _space_id: String, _server_id: String) -> Result<Value> {
-        Err(gateway_write_unavailable())
+    async fn retry_connection(&self, space_id: String, server_id: String) -> Result<Value> {
+        let space_uuid = Uuid::parse_str(&space_id)?;
+
+        self.pool_service.remove_instance(space_uuid, &server_id);
+
+        let installed = self
+            .installed_server_repo
+            .get_by_server_id(&space_id, &server_id)
+            .await?
+            .ok_or_else(|| anyhow!("Server not found: {space_id}/{server_id}"))?;
+
+        let server_definition = installed
+            .get_definition()
+            .ok_or_else(|| anyhow!("Server {server_id} has no cached definition"))?;
+
+        let key = ServerKey::new(space_uuid, &server_id);
+        self.server_manager.set_connecting(&key).await;
+
+        let transport = build_transport_config(
+            &server_definition.transport,
+            &installed,
+            Some(&self.data_dir),
+            TransportResolutionOptions::default(),
+        );
+
+        let ctx = ConnectionContext::auto(space_uuid, server_id.clone(), transport);
+        let result = self.pool_service.connect_server(&ctx).await;
+
+        match result {
+            ConnectionResult::Connected { features, .. } => {
+                self.server_manager.set_connected(&key, features).await;
+            }
+            ConnectionResult::OAuthRequired { .. } => {
+                self.server_manager.set_auth_required(&key, None).await;
+                if let Err(error) = self
+                    .feature_service
+                    .mark_unavailable(&space_id, &server_id)
+                    .await
+                {
+                    warn!("[LiveGatewayWriteRuntime] Failed to mark features unavailable: {error}");
+                }
+            }
+            ConnectionResult::Failed { error } => {
+                self.server_manager.set_error(&key, error.clone()).await;
+                if let Err(mark_error) = self
+                    .feature_service
+                    .mark_unavailable(&space_id, &server_id)
+                    .await
+                {
+                    warn!(
+                        "[LiveGatewayWriteRuntime] Failed to mark features unavailable: {mark_error}"
+                    );
+                }
+                return Err(anyhow!(error));
+            }
+        }
+
+        Ok(json!({ "ok": true }))
     }
 
     async fn update_server_package(&self, space_id: String, server_id: String) -> Result<Value> {

crates/mcpmux-gateway/src/consumers/mod.rs

@@ -31,6 +31,8 @@
 
 mod mcp_notifier;
 mod oauth_handler;
+mod server_config_handler;
 
 pub use mcp_notifier::MCPNotifier;
 pub use oauth_handler::OAuthEventHandler;
+pub use server_config_handler::ServerConfigUpdatedHandler;

crates/mcpmux-gateway/src/consumers/server_config_handler.rs

@@ -0,0 +1,92 @@
+//! Server Config Event Handler - Evicts pool instances on config changes
+//!
+//! Listens for `ServerConfigUpdated` and removes the pooled instance for
+//! enabled servers so the next connect rebuilds transport from DB.
+
+use mcpmux_core::{DomainEvent, InstalledServerRepository};
+use std::sync::Arc;
+use tokio::sync::broadcast;
+use tracing::{debug, info, warn};
+use uuid::Uuid;
+
+use crate::pool::PoolService;
+
+/// Evicts pooled server instances when configuration changes.
+pub struct ServerConfigUpdatedHandler {
+    installed_server_repo: Arc<dyn InstalledServerRepository + Send + Sync>,
+    pool_service: Arc<PoolService>,
+}
+
+impl ServerConfigUpdatedHandler {
+    /// Create a handler wired to the installed-server repo and connection pool.
+    pub fn new(
+        installed_server_repo: Arc<dyn InstalledServerRepository + Send + Sync>,
+        pool_service: Arc<PoolService>,
+    ) -> Self {
+        Self {
+            installed_server_repo,
+            pool_service,
+        }
+    }
+
+    /// Start listening to domain events on a background task.
+    pub fn start(self: Arc<Self>, mut event_rx: broadcast::Receiver<DomainEvent>) {
+        tokio::spawn(async move {
+            info!("[ServerConfigHandler] Started listening for ServerConfigUpdated events");
+
+            loop {
+                match event_rx.recv().await {
+                    Ok(event) => {
+                        if let Err(error) = self.handle_event(event).await {
+                            warn!("[ServerConfigHandler] Failed to handle event: {error}");
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                        warn!("[ServerConfigHandler] Lagged behind, skipped {skipped} events");
+                    }
+                    Err(broadcast::error::RecvError::Closed) => {
+                        warn!("[ServerConfigHandler] Event channel closed");
+                        break;
+                    }
+                }
+            }
+        });
+    }
+
+    /// Handle one domain event, evicting the pool instance when applicable.
+    async fn handle_event(&self, event: DomainEvent) -> anyhow::Result<()> {
+        let DomainEvent::ServerConfigUpdated { space_id, server_id } = event else {
+            return Ok(());
+        };
+
+        self.handle_config_updated(space_id, &server_id).await
+    }
+
+    /// Remove a pooled instance for an enabled server after a config/definition write.
+    async fn handle_config_updated(&self, space_id: Uuid, server_id: &str) -> anyhow::Result<()> {
+        let space_id_str = space_id.to_string();
+        let Some(installed) = self
+            .installed_server_repo
+            .get_by_server_id(&space_id_str, server_id)
+            .await?
+        else {
+            debug!(
+                "[ServerConfigHandler] Server {space_id}/{server_id} not found, skipping eviction"
+            );
+            return Ok(());
+        };
+
+        if !installed.enabled {
+            debug!(
+                "[ServerConfigHandler] Skipping eviction for disabled server {space_id}/{server_id}"
+            );
+            return Ok(());
+        }
+
+        info!(
+            "[ServerConfigHandler] Evicting pool instance for enabled server {space_id}/{server_id} after config update"
+        );
+        self.pool_service.remove_instance(space_id, server_id);
+        Ok(())
+    }
+}