fix(core): avoid holding pool registry locks during database I/O

crates/dbx-core/src/agent_explain.rs

@@ -38,8 +38,8 @@ pub async fn get_agent_explain_info_core(
     }
 
     let target = {
-        let connections = state.connections.read().await;
-        let pool = connections.get(&pool_key).ok_or_else(|| "Connection not found".to_string())?;
+        let pool_handle = state.pool_handle(&pool_key).await;
+        let pool = pool_handle.as_ref().ok_or_else(|| "Connection not found".to_string())?;
         match pool {
             PoolKind::Agent(client) => ExplainTarget::Agent(client.clone()),
             PoolKind::ExternalDriver { config, session, .. } => {
@@ -227,14 +227,18 @@ mod tests {
         let storage = crate::storage::Storage::open(&dir.join("storage.db")).await.unwrap();
         let state = AppState::new(storage);
         state.configs.write().await.insert(config.id.clone(), config.clone());
-        state.connections.write().await.insert(
-            "oracle-jdbc".to_string(),
-            PoolKind::ExternalDriver {
-                driver_id: "jdbc".to_string(),
-                config: Arc::new(config),
-                session: session.clone(),
-            },
-        );
+        state
+            .update_connection_pools(|connections| {
+                connections.insert(
+                    "oracle-jdbc".to_string(),
+                    PoolKind::ExternalDriver {
+                        driver_id: "jdbc".to_string(),
+                        config: Arc::new(config),
+                        session: session.clone(),
+                    },
+                );
+            })
+            .await;
 
         let plan = get_agent_explain_info_core(
             &state,

crates/dbx-core/src/agent_kv.rs

@@ -1013,8 +1013,8 @@ async fn ensure_etcd_dangerous_action_capability(
             .get(connection_id)
             .and_then(|config| crate::agent_catalog::agent_key(&config.db_type, config.driver_profile.as_deref()))
     };
-    let connections = state.connections.read().await;
-    let PoolKind::Agent(client) = connections.get(connection_id).ok_or("Connection not found")? else {
+    let pool = state.pool_handle(connection_id).await.ok_or("Connection not found")?;
+    let PoolKind::Agent(client) = &pool else {
         return Err("Not an agent key-value connection".to_string());
     };
     if !client.lock().await.supports_capability(capability) {
@@ -1706,8 +1706,8 @@ fn kv_put_required_capabilities(options: &KvPutOptions) -> Vec<AgentCapability>
 
 pub async fn kv_supports_ttl_core(state: &AppState, connection_id: &str) -> Result<bool, String> {
     ensure_agent_kv_pool(state, connection_id).await?;
-    let connections = state.connections.read().await;
-    let pool = connections.get(connection_id).ok_or("Connection not found")?;
+    let pool_handle = state.pool_handle(connection_id).await;
+    let pool = pool_handle.as_ref().ok_or("Connection not found")?;
     match pool {
         PoolKind::Agent(client) => Ok(client.lock().await.supports_capability(AgentCapability::KvTtl)),
         _ => Err("Not an agent key-value connection".to_string()),
@@ -1738,8 +1738,8 @@ async fn call_agent_kv<T: serde::de::DeserializeOwned + Send + 'static>(
             .and_then(|config| crate::agent_catalog::agent_key(&config.db_type, config.driver_profile.as_deref()))
     };
     let client = {
-        let connections = state.connections.read().await;
-        match connections.get(connection_id) {
+        let pool_handle = state.pool_handle(connection_id).await;
+        match pool_handle.as_ref() {
             Some(PoolKind::Agent(client)) => client.clone(),
             Some(_) => return Err("Not an agent key-value connection".to_string()),
             None => return Err("Connection not found".to_string()),

crates/dbx-core/src/agent_tools.rs

@@ -1872,7 +1872,11 @@ for line in sys.stdin:
         let state = Arc::new(AppState::new(storage));
         let connection = agent_test_connection("dameng-1", "Dameng", DatabaseType::Dameng, "APPDB");
         state.configs.write().await.insert(connection.id.clone(), connection);
-        state.connections.write().await.insert("dameng-1:APPDB".to_string(), PoolKind::agent(client));
+        state
+            .update_connection_pools(|connections| {
+                connections.insert("dameng-1:APPDB".to_string(), PoolKind::agent(client));
+            })
+            .await;
 
         let read = ToolCall {
             id: "read".to_string(),
@@ -1936,7 +1940,11 @@ for line in sys.stdin:
         let state = Arc::new(AppState::new(storage));
         let connection = agent_test_connection("mysql-1", "MySQL", DatabaseType::Mysql, "rs_main");
         state.configs.write().await.insert(connection.id.clone(), connection);
-        state.connections.write().await.insert("mysql-1:rs_main".to_string(), PoolKind::agent(client));
+        state
+            .update_connection_pools(|connections| {
+                connections.insert("mysql-1:rs_main".to_string(), PoolKind::agent(client));
+            })
+            .await;
 
         let sql = "SHOW TRIGGERS FROM `rs_main` LIKE 'trg_order_items_after_%';";
         let call = ToolCall {