feat(agent): pluggable database drivers via JDBC agent subprocess

Migrate domestic database support (DaMeng, KingBase, Vastbase, GoldenDB) from
built-in ODBC/protocol drivers to standalone Java agent processes communicating
via stdin/stdout JSON-RPC 2.0. Agents are distributed as fat JARs with embedded
JDBC drivers, downloaded on-demand from GitHub Releases with proxy fallback.

- Add AgentDriverClient for subprocess lifecycle + JSON-RPC communication
- Add AgentManager for JRE/JAR install state management
- Add Tauri commands for driver install/uninstall/list
- Add DriverManager UI panel in settings dialog
- Extend DatabaseType with Kingbase, Vastbase, Goldendb variants
- Migrate Dameng from ODBC (dm_driver.rs) to JDBC agent
- Delete dm_driver.rs (343 lines)
- Add GitHub proxy fallback for downloads (update.hwdns.net, gh-proxy.org)

Agent code lives at https://github.com/t8y2/dbx-agents

crates/dbx-core/src/agent_manager.rs

@@ -0,0 +1,172 @@
+use std::path::{Path, PathBuf};
+
+use serde::{Deserialize, Serialize};
+
+use crate::db::agent_driver::AgentDriverClient;
+use crate::models::connection::DatabaseType;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentRegistry {
+    pub jre: JreInfo,
+    pub drivers: std::collections::HashMap<String, DriverInfo>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JreInfo {
+    pub version: String,
+    pub platforms: std::collections::HashMap<String, ArtifactInfo>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DriverInfo {
+    pub version: String,
+    pub label: String,
+    pub min_app_version: String,
+    pub jar: ArtifactInfo,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ArtifactInfo {
+    pub url: String,
+    pub sha256: String,
+    pub size: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentState {
+    #[serde(default)]
+    pub jre_version: Option<String>,
+    #[serde(default)]
+    pub installed_drivers: std::collections::HashMap<String, InstalledDriver>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InstalledDriver {
+    pub version: String,
+    pub installed_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentDriverInfo {
+    pub db_type: String,
+    pub label: String,
+    pub version: String,
+    pub size: u64,
+    pub installed: bool,
+    pub installed_version: Option<String>,
+    pub update_available: bool,
+}
+
+pub struct AgentManager {
+    base_dir: PathBuf,
+}
+
+impl AgentManager {
+    pub fn new() -> Self {
+        let home =
+            std::env::var(if cfg!(windows) { "USERPROFILE" } else { "HOME" }).unwrap_or_else(|_| ".".to_string());
+        Self { base_dir: PathBuf::from(home).join(".dbx").join("agents") }
+    }
+
+    pub fn base_dir(&self) -> &PathBuf {
+        &self.base_dir
+    }
+
+    pub fn jre_java_path(&self) -> PathBuf {
+        if cfg!(windows) {
+            self.base_dir.join("jre").join("bin").join("java.exe")
+        } else {
+            self.base_dir.join("jre").join("bin").join("java")
+        }
+    }
+
+    pub fn driver_jar_path(&self, db_type: &str) -> PathBuf {
+        self.base_dir.join("drivers").join(db_type).join("agent.jar")
+    }
+
+    fn state_path(&self) -> PathBuf {
+        self.base_dir.join("state.json")
+    }
+
+    pub fn load_state(&self) -> AgentState {
+        std::fs::read_to_string(self.state_path())
+            .ok()
+            .and_then(|s| serde_json::from_str(&s).ok())
+            .unwrap_or(AgentState { jre_version: None, installed_drivers: Default::default() })
+    }
+
+    pub fn save_state(&self, state: &AgentState) -> Result<(), String> {
+        let dir = self.base_dir.clone();
+        std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
+        let json = serde_json::to_string_pretty(state).map_err(|e| e.to_string())?;
+        std::fs::write(self.state_path(), json).map_err(|e| e.to_string())
+    }
+
+    pub fn is_jre_installed(&self) -> bool {
+        self.jre_java_path().exists()
+    }
+
+    pub fn is_driver_installed(&self, db_type: &str) -> bool {
+        self.driver_jar_path(db_type).exists()
+    }
+
+    pub fn db_type_to_agent_key(db_type: &DatabaseType) -> Option<&'static str> {
+        match db_type {
+            DatabaseType::Dameng => Some("dameng"),
+            DatabaseType::Kingbase => Some("kingbase"),
+            DatabaseType::Vastbase => Some("vastbase"),
+            DatabaseType::Goldendb => Some("goldendb"),
+            _ => None,
+        }
+    }
+
+    pub fn is_agent_type(db_type: &DatabaseType) -> bool {
+        Self::db_type_to_agent_key(db_type).is_some()
+    }
+
+    pub async fn spawn(&self, db_type: &DatabaseType) -> Result<AgentDriverClient, String> {
+        let key = Self::db_type_to_agent_key(db_type)
+            .ok_or_else(|| format!("{:?} is not an agent-driven database type", db_type))?;
+
+        if !self.is_jre_installed() {
+            return Err("JRE runtime is not installed. Please install it from the Driver Manager.".to_string());
+        }
+        if !self.is_driver_installed(key) {
+            return Err(format!("{key} driver is not installed. Please install it from the Driver Manager."));
+        }
+
+        let java = self.jre_java_path().to_string_lossy().to_string();
+        let jar = self.driver_jar_path(key).to_string_lossy().to_string();
+        AgentDriverClient::spawn(&java, &jar).await
+    }
+
+    pub async fn download_file(url: &str, dest: &Path) -> Result<(), String> {
+        if let Some(parent) = dest.parent() {
+            std::fs::create_dir_all(parent).map_err(|e| e.to_string())?;
+        }
+        let resp = reqwest::get(url).await.map_err(|e| format!("Download failed: {e}"))?;
+        if !resp.status().is_success() {
+            return Err(format!("Download failed with status: {}", resp.status()));
+        }
+        let bytes = resp.bytes().await.map_err(|e| format!("Download read failed: {e}"))?;
+        std::fs::write(dest, &bytes).map_err(|e| format!("Failed to write file: {e}"))
+    }
+
+    pub fn current_platform() -> &'static str {
+        if cfg!(target_os = "macos") && cfg!(target_arch = "aarch64") {
+            "macos-aarch64"
+        } else if cfg!(target_os = "macos") && cfg!(target_arch = "x86_64") {
+            "macos-x64"
+        } else if cfg!(target_os = "linux") && cfg!(target_arch = "aarch64") {
+            "linux-aarch64"
+        } else if cfg!(target_os = "linux") && cfg!(target_arch = "x86_64") {
+            "linux-x64"
+        } else if cfg!(target_os = "windows") && cfg!(target_arch = "aarch64") {
+            "windows-aarch64"
+        } else if cfg!(target_os = "windows") && cfg!(target_arch = "x86_64") {
+            "windows-x64"
+        } else {
+            "unknown"
+        }
+    }
+}

crates/dbx-core/src/connection.rs

@@ -45,7 +45,7 @@ pub enum PoolKind {
     SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
     Oracle(Arc<OraclePool>),
     Elasticsearch(db::elasticsearch_driver::EsClient),
-    Dameng(Arc<std::sync::Mutex<db::dm_driver::DmClient>>),
+    Agent(Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>),
     Gaussdb(Arc<tokio::sync::Mutex<db::gaussdb_driver::GaussdbClient>>),
     ExternalTabular(Arc<external::ExternalPool>),
     ExternalDriver { driver_id: String, config: ConnectionConfig, session: Arc<PluginDriverSession> },
@@ -99,6 +99,7 @@ pub struct AppState {
     pub proxy_tunnels: ProxyTunnelManager,
     pub storage: Storage,
     pub plugins: PluginRegistry,
+    pub agent_manager: crate::agent_manager::AgentManager,
 }
 
 pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig {
@@ -112,7 +113,7 @@ pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig
 pub fn database_connection_config(config: &ConnectionConfig, database: Option<&str>) -> ConnectionConfig {
     let mut db_config = if database.is_some() { config.clone() } else { metadata_connection_config(config) };
     if let Some(db) = database {
-        if db_config.db_type != DatabaseType::Oracle && db_config.db_type != DatabaseType::Dameng {
+        if !matches!(db_config.db_type, DatabaseType::Oracle | DatabaseType::Dameng) {
             db_config.database = Some(db.to_string());
         }
     }
@@ -133,6 +134,7 @@ impl AppState {
             proxy_tunnels: ProxyTunnelManager::new(),
             storage,
             plugins: PluginRegistry::new(plugin_dir),
+            agent_manager: crate::agent_manager::AgentManager::new(),
         }
     }
 
@@ -169,6 +171,9 @@ impl AppState {
                 | Some(DatabaseType::DuckDb)
                 | Some(DatabaseType::Oracle)
                 | Some(DatabaseType::Dameng)
+                | Some(DatabaseType::Kingbase)
+                | Some(DatabaseType::Vastbase)
+                | Some(DatabaseType::Goldendb)
                 | Some(DatabaseType::Jdbc)
         );
         let pool_key = if is_single_conn {
@@ -272,16 +277,21 @@ impl AppState {
                 db::elasticsearch_driver::test_connection(&client).await?;
                 PoolKind::Elasticsearch(client)
             }
-            DatabaseType::Dameng => {
-                let client = db::dm_driver::connect(
-                    &host,
-                    port,
-                    db_config.database.as_deref().unwrap_or(""),
-                    &db_config.username,
-                    &db_config.password,
-                )
-                .await?;
-                PoolKind::Dameng(Arc::new(std::sync::Mutex::new(client)))
+            DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => {
+                let mut client = self.agent_manager.spawn(&db_config.db_type).await?;
+                client
+                    .call::<serde_json::Value>(
+                        "connect",
+                        serde_json::json!({
+                            "host": host,
+                            "port": port,
+                            "database": db_config.effective_database().unwrap_or(""),
+                            "username": db_config.username,
+                            "password": db_config.password,
+                        }),
+                    )
+                    .await?;
+                PoolKind::Agent(Arc::new(tokio::sync::Mutex::new(client)))
             }
             DatabaseType::Gaussdb => {
                 let client = db::gaussdb_driver::connect(
@@ -404,9 +414,15 @@ impl AppState {
             configs
                 .get(connection_id)
                 .map(|c| {
-                    c.db_type == DatabaseType::Oracle
-                        || c.db_type == DatabaseType::Elasticsearch
-                        || c.db_type == DatabaseType::Dameng
+                    matches!(
+                        c.db_type,
+                        DatabaseType::Oracle
+                            | DatabaseType::Elasticsearch
+                            | DatabaseType::Dameng
+                            | DatabaseType::Kingbase
+                            | DatabaseType::Vastbase
+                            | DatabaseType::Goldendb
+                    )
                 })
                 .unwrap_or(false)
         };
@@ -449,6 +465,7 @@ pub async fn probe_connection_endpoint(config: &ConnectionConfig, host: &str, po
         DatabaseType::Sqlite | DatabaseType::DuckDb => Ok(()),
         DatabaseType::MongoDb if config.connection_string.as_deref().is_some_and(|value| !value.is_empty()) => Ok(()),
         DatabaseType::Jdbc => Ok(()),
+        DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => Ok(()),
         _ => db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await,
     }
 }

crates/dbx-core/src/db/agent_driver.rs

@@ -0,0 +1,162 @@
+use std::io::{BufRead, BufReader, BufWriter, Write};
+use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
+use std::time::Duration;
+
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+const RPC_TIMEOUT_SECS: u64 = 30;
+const STARTUP_TIMEOUT_SECS: u64 = 15;
+
+pub struct AgentDriverClient {
+    child: Child,
+    stdin: Option<BufWriter<ChildStdin>>,
+    stdout: Option<BufReader<ChildStdout>>,
+    next_id: u64,
+}
+
+impl AgentDriverClient {
+    /// Spawn a Java agent process and wait for it to signal readiness.
+    ///
+    /// The agent is started via `java -jar <jar_path>` with stdin/stdout piped.
+    /// Blocks (async) until the agent writes `{"ready":true}` to stdout.
+    pub async fn spawn(java_path: &str, jar_path: &str) -> Result<Self, String> {
+        let mut child = Command::new(java_path)
+            .arg("-jar")
+            .arg(jar_path)
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .map_err(|e| format!("Failed to spawn agent process: {e}"))?;
+
+        let child_stdin = child.stdin.take().ok_or("Failed to capture agent stdin")?;
+        let child_stdout = child.stdout.take().ok_or("Failed to capture agent stdout")?;
+
+        let stdin = BufWriter::new(child_stdin);
+        let mut stdout = BufReader::new(child_stdout);
+
+        // Wait for the agent to signal readiness with {"ready":true}
+        let ready_stdout = tokio::time::timeout(
+            Duration::from_secs(STARTUP_TIMEOUT_SECS),
+            tokio::task::spawn_blocking(move || {
+                let mut line = String::new();
+                stdout.read_line(&mut line).map_err(|e| format!("Failed to read startup line from agent: {e}"))?;
+                let v: Value = serde_json::from_str(line.trim())
+                    .map_err(|e| format!("Invalid JSON from agent during startup: {e}"))?;
+                if v.get("ready") != Some(&Value::Bool(true)) {
+                    return Err(format!("Agent did not send ready signal, got: {line}"));
+                }
+                Ok(stdout)
+            }),
+        )
+        .await
+        .map_err(|_| format!("Agent startup timed out ({STARTUP_TIMEOUT_SECS}s)"))?
+        .map_err(|e| format!("Agent startup task failed: {e}"))??;
+
+        Ok(Self { child, stdin: Some(stdin), stdout: Some(ready_stdout), next_id: 0 })
+    }
+
+    /// Send a JSON-RPC 2.0 request and wait for the response.
+    pub async fn call<T: DeserializeOwned + Send + 'static>(
+        &mut self,
+        method: &str,
+        params: Value,
+    ) -> Result<T, String> {
+        self.next_id += 1;
+        let id = self.next_id;
+
+        let request = serde_json::json!({
+            "jsonrpc": "2.0",
+            "id": id,
+            "method": method,
+            "params": params,
+        });
+        let request_line =
+            serde_json::to_string(&request).map_err(|e| format!("Failed to serialize JSON-RPC request: {e}"))?;
+
+        // Write request to stdin
+        {
+            let writer = self.stdin.as_mut().ok_or("Agent stdin not available")?;
+            writer.write_all(request_line.as_bytes()).map_err(|e| format!("Failed to write to agent stdin: {e}"))?;
+            writer.write_all(b"\n").map_err(|e| format!("Failed to write newline to agent stdin: {e}"))?;
+            writer.flush().map_err(|e| format!("Failed to flush agent stdin: {e}"))?;
+        }
+
+        // Read response from stdout (blocking, with timeout)
+        let mut reader = self.stdout.take().ok_or("Agent stdout not available")?;
+
+        let (returned_reader, result) = tokio::time::timeout(
+            Duration::from_secs(RPC_TIMEOUT_SECS),
+            tokio::task::spawn_blocking(move || {
+                let mut line = String::new();
+                let read_result =
+                    reader.read_line(&mut line).map_err(|e| format!("Failed to read response from agent: {e}"));
+                if let Err(e) = read_result {
+                    return (reader, Err(e));
+                }
+
+                let resp: Value = match serde_json::from_str(line.trim()) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        return (reader, Err(format!("Invalid JSON response from agent: {e}")));
+                    }
+                };
+
+                let result = if let Some(err) = resp.get("error") {
+                    let msg = err.get("message").and_then(|m| m.as_str()).unwrap_or("Unknown agent error");
+                    let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(-1);
+                    Err(format!("Agent RPC error ({code}): {msg}"))
+                } else if let Some(result_val) = resp.get("result") {
+                    serde_json::from_value::<T>(result_val.clone())
+                        .map_err(|e| format!("Failed to deserialize agent result: {e}"))
+                } else {
+                    Err(format!("Agent response missing both 'result' and 'error': {line}"))
+                };
+
+                (reader, result)
+            }),
+        )
+        .await
+        .map_err(|_| format!("Agent RPC call timed out ({RPC_TIMEOUT_SECS}s)"))?
+        .map_err(|e| format!("Agent RPC task failed: {e}"))?;
+
+        let _ = self.stdout.insert(returned_reader);
+        result
+    }
+
+    /// Send a shutdown message to the agent and wait for the process to exit.
+    pub async fn shutdown(&mut self) {
+        // Try to send a shutdown RPC; ignore errors if the agent is already gone
+        let shutdown_result: Result<Value, String> = self.call("shutdown", Value::Null).await;
+        if let Err(e) = &shutdown_result {
+            log::warn!("Agent shutdown RPC failed: {e}");
+        }
+
+        // Drop stdin to signal EOF
+        self.stdin.take();
+
+        // Wait for the child to exit
+        match self.child.wait() {
+            Ok(status) => log::info!("Agent process exited with {status}"),
+            Err(e) => log::warn!("Failed to wait for agent process: {e}"),
+        }
+    }
+
+    /// Forcefully kill the agent process.
+    pub fn kill(&mut self) {
+        self.stdin.take();
+        self.stdout.take();
+        if let Err(e) = self.child.kill() {
+            log::warn!("Failed to kill agent process: {e}");
+        }
+        // Reap the child to avoid zombie processes
+        let _ = self.child.wait();
+    }
+}
+
+impl Drop for AgentDriverClient {
+    fn drop(&mut self) {
+        self.kill();
+    }
+}

crates/dbx-core/src/db/dm_driver.rs

@@ -1,343 +0,0 @@
-use std::time::Instant;
-
-use odbc_api::{buffers::TextRowSet, ConnectionOptions, Cursor, ResultSetMetadata};
-
-use crate::sql::starts_with_executable_sql_keyword;
-use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
-
-use super::CONNECTION_TIMEOUT_SECS;
-
-pub struct DmClient {
-    conn: odbc_api::Connection<'static>,
-}
-
-unsafe impl Send for DmClient {}
-
-impl DmClient {
-    pub fn query_rows(&self, sql: &str) -> Result<Vec<Vec<String>>, String> {
-        match self.conn.execute(sql, (), None).map_err(|e| e.to_string())? {
-            Some(cursor) => read_cursor(cursor),
-            None => Ok(vec![]),
-        }
-    }
-
-    fn query_single_column(&self, sql: &str) -> Result<Vec<String>, String> {
-        Ok(self.query_rows(sql)?.into_iter().filter_map(|r| r.into_iter().next()).collect())
-    }
-}
-
-fn read_cursor(cursor: impl Cursor) -> Result<Vec<Vec<String>>, String> {
-    let mut cursor = cursor;
-    let col_count = cursor.num_result_cols().map_err(|e| e.to_string())? as u16;
-    let buffer = TextRowSet::for_cursor(1000, &mut cursor, Some(8192)).map_err(|e| e.to_string())?;
-    let mut row_cursor = cursor.bind_buffer(buffer).map_err(|e| e.to_string())?;
-    let mut rows = Vec::new();
-    while let Some(batch) = row_cursor.fetch().map_err(|e| e.to_string())? {
-        for row_idx in 0..batch.num_rows() {
-            let vals: Vec<String> = (0..col_count as usize)
-                .map(|col| {
-                    batch.at(col, row_idx).and_then(|bytes| std::str::from_utf8(bytes).ok()).unwrap_or("").to_string()
-                })
-                .collect();
-            rows.push(vals);
-        }
-    }
-    Ok(rows)
-}
-
-pub async fn connect(host: &str, port: u16, database: &str, user: &str, pass: &str) -> Result<DmClient, String> {
-    let conn_str = format!(
-        "Driver={{DM8 ODBC DRIVER}};Server={host};TCP_PORT={port};DATABASE={db};UID={user};PWD={pass}",
-        host = host,
-        port = port,
-        db = database,
-        user = user,
-        pass = pass,
-    );
-
-    let result = tokio::time::timeout(
-        std::time::Duration::from_secs(CONNECTION_TIMEOUT_SECS),
-        tokio::task::spawn_blocking(move || {
-            super::ODBC_ENV.connect_with_connection_string(&conn_str, ConnectionOptions::default())
-                .map_err(|e| {
-                    let msg = e.to_string();
-                    if msg.contains("Data source name not found") || msg.contains("Can't open lib") {
-                        format!(
-                            "DM8 ODBC driver not found. Please install the DM8 ODBC driver \
-                             and register it in odbcinst.ini (Linux/macOS) or the ODBC Data Source Administrator (Windows). \
-                             Original error: {msg}"
-                        )
-                    } else {
-                        format!("DM connection failed: {msg}")
-                    }
-                })
-                .map(|conn| DmClient { conn })
-        }),
-    )
-    .await
-    .map_err(|_| format!("DM connection timed out ({CONNECTION_TIMEOUT_SECS}s)"))?
-    .map_err(|e| format!("DM connection task failed: {e}"))?;
-
-    result
-}
-
-pub fn list_databases(client: &DmClient) -> Result<Vec<DatabaseInfo>, String> {
-    let rows = client.query_single_column(
-        "SELECT USERNAME FROM ALL_USERS \
-         WHERE USERNAME NOT IN (\
-           'SYS','SYSDBA','SYSAUDITOR','SYSSSO','CTISYS',\
-           'SYS_DBA','_SYS_STATISTICS','SYS_PHM'\
-         ) ORDER BY USERNAME",
-    )?;
-    Ok(rows.into_iter().map(|name| DatabaseInfo { name }).collect())
-}
-
-pub fn list_schemas(client: &DmClient) -> Result<Vec<String>, String> {
-    let dbs = list_databases(client)?;
-    Ok(dbs.into_iter().map(|d| d.name).collect())
-}
-
-pub fn list_tables(client: &DmClient, schema: &str) -> Result<Vec<TableInfo>, String> {
-    let s = schema.replace('\'', "''");
-    let sql = format!(
-        "SELECT TABLE_NAME, 'TABLE' AS TABLE_TYPE FROM ALL_TABLES WHERE OWNER = '{s}' \
-         UNION ALL \
-         SELECT VIEW_NAME, 'VIEW' FROM ALL_VIEWS WHERE OWNER = '{s}' \
-         ORDER BY 1"
-    );
-    let rows = client.query_rows(&sql)?;
-    Ok(rows
-        .into_iter()
-        .map(|r| TableInfo {
-            name: r.first().cloned().unwrap_or_default(),
-            table_type: r.get(1).cloned().unwrap_or_else(|| "TABLE".to_string()),
-            comment: None,
-        })
-        .collect())
-}
-
-pub fn get_columns(client: &DmClient, schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
-    let s = schema.replace('\'', "''");
-    let t = table.replace('\'', "''");
-
-    let pk_rows = client.query_single_column(&format!(
-        "SELECT cols.COLUMN_NAME FROM ALL_CONS_COLUMNS cols \
-         JOIN ALL_CONSTRAINTS cons ON cols.CONSTRAINT_NAME = cons.CONSTRAINT_NAME AND cols.OWNER = cons.OWNER \
-         WHERE cons.CONSTRAINT_TYPE = 'P' AND cons.OWNER = '{s}' AND cons.TABLE_NAME = '{t}'"
-    ))?;
-    let pk_names: std::collections::HashSet<String> = pk_rows.into_iter().collect();
-
-    let col_rows = client.query_rows(&format!(
-        "SELECT COLUMN_NAME, DATA_TYPE, NULLABLE, DATA_PRECISION, DATA_SCALE, DATA_LENGTH, CHAR_LENGTH \
-         FROM ALL_TAB_COLUMNS \
-         WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \
-         ORDER BY COLUMN_ID"
-    ))?;
-
-    Ok(col_rows
-        .into_iter()
-        .map(|r| {
-            let name = r.first().cloned().unwrap_or_default();
-            let base = r.get(1).cloned().unwrap_or_default();
-            let num_prec = r.get(3).and_then(|v| v.parse::<i32>().ok());
-            let num_scale = r.get(4).and_then(|v| v.parse::<i32>().ok());
-            let data_len = r.get(5).and_then(|v| v.parse::<i32>().ok());
-            let char_len = r.get(6).and_then(|v| v.parse::<i32>().ok());
-            let data_type = match base.to_uppercase().as_str() {
-                "VARCHAR2" | "NVARCHAR2" | "VARCHAR" | "CHAR" | "NCHAR" => {
-                    let len = char_len.or(data_len);
-                    match len {
-                        Some(n) => format!("{base}({n})"),
-                        None => base,
-                    }
-                }
-                "NUMBER" | "NUMERIC" | "DECIMAL" => match (num_prec, num_scale) {
-                    (Some(p), Some(s)) if s > 0 => format!("{base}({p},{s})"),
-                    (Some(p), _) if p > 0 => format!("{base}({p})"),
-                    _ => base,
-                },
-                "RAW" => match data_len {
-                    Some(n) => format!("RAW({n})"),
-                    None => "RAW".to_string(),
-                },
-                _ => base,
-            };
-            ColumnInfo {
-                is_primary_key: pk_names.contains(&name),
-                name,
-                data_type,
-                is_nullable: r.get(2).map(|v| v == "Y").unwrap_or(false),
-                column_default: None,
-                extra: None,
-                comment: None,
-                numeric_precision: num_prec,
-                numeric_scale: num_scale,
-                character_maximum_length: char_len,
-            }
-        })
-        .collect())
-}
-
-pub fn list_indexes(client: &DmClient, schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
-    let s = schema.replace('\'', "''");
-    let t = table.replace('\'', "''");
-    let sql = format!(
-        "SELECT i.INDEX_NAME, \
-         LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS COLUMNS, \
-         i.UNIQUENESS, \
-         CASE WHEN c.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PK, \
-         i.INDEX_TYPE \
-         FROM ALL_INDEXES i \
-         JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER \
-         LEFT JOIN ALL_CONSTRAINTS c ON i.INDEX_NAME = c.INDEX_NAME AND i.TABLE_OWNER = c.OWNER \
-           AND c.CONSTRAINT_TYPE = 'P' \
-         WHERE i.TABLE_OWNER = '{s}' AND i.TABLE_NAME = '{t}' \
-         GROUP BY i.INDEX_NAME, i.UNIQUENESS, c.CONSTRAINT_TYPE, i.INDEX_TYPE \
-         ORDER BY i.INDEX_NAME"
-    );
-    let rows = client.query_rows(&sql)?;
-    Ok(rows
-        .into_iter()
-        .map(|r| {
-            let cols_str = r.get(1).cloned().unwrap_or_default();
-            IndexInfo {
-                name: r.first().cloned().unwrap_or_default(),
-                columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(),
-                is_unique: r.get(2).map(|v| v == "UNIQUE").unwrap_or(false),
-                is_primary: r.get(3).map(|v| v == "1").unwrap_or(false),
-                filter: None,
-                index_type: r.get(4).cloned(),
-                included_columns: None,
-                comment: None,
-            }
-        })
-        .collect())
-}
-
-pub fn list_foreign_keys(client: &DmClient, schema: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
-    let s = schema.replace('\'', "''");
-    let t = table.replace('\'', "''");
-    let sql = format!(
-        "SELECT c.CONSTRAINT_NAME, cc.COLUMN_NAME, rc.TABLE_NAME, rcc.COLUMN_NAME \
-         FROM ALL_CONSTRAINTS c \
-         JOIN ALL_CONS_COLUMNS cc ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME AND c.OWNER = cc.OWNER \
-         JOIN ALL_CONSTRAINTS rc ON c.R_CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND c.R_OWNER = rc.OWNER \
-         JOIN ALL_CONS_COLUMNS rcc ON rc.CONSTRAINT_NAME = rcc.CONSTRAINT_NAME AND rc.OWNER = rcc.OWNER \
-         WHERE c.CONSTRAINT_TYPE = 'R' AND c.OWNER = '{s}' AND c.TABLE_NAME = '{t}' \
-         ORDER BY c.CONSTRAINT_NAME"
-    );
-    let rows = client.query_rows(&sql)?;
-    Ok(rows
-        .into_iter()
-        .map(|r| ForeignKeyInfo {
-            name: r.first().cloned().unwrap_or_default(),
-            column: r.get(1).cloned().unwrap_or_default(),
-            ref_table: r.get(2).cloned().unwrap_or_default(),
-            ref_column: r.get(3).cloned().unwrap_or_default(),
-        })
-        .collect())
-}
-
-pub fn list_triggers(client: &DmClient, schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
-    let s = schema.replace('\'', "''");
-    let t = table.replace('\'', "''");
-    let sql = format!(
-        "SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \
-         FROM ALL_TRIGGERS \
-         WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \
-         ORDER BY TRIGGER_NAME"
-    );
-    let rows = client.query_rows(&sql)?;
-    Ok(rows
-        .into_iter()
-        .map(|r| TriggerInfo {
-            name: r.first().cloned().unwrap_or_default(),
-            event: r.get(1).cloned().unwrap_or_default(),
-            timing: r.get(2).cloned().unwrap_or_default(),
-        })
-        .collect())
-}
-
-pub fn execute_query_with_schema_sync(client: &DmClient, schema: &str, sql: &str) -> Result<QueryResult, String> {
-    let set_schema = format!("SET SCHEMA \"{}\"", schema);
-    client.conn.execute(&set_schema, (), None).map_err(|e| {
-        log::error!("[dameng] set schema failed: {e}");
-        e.to_string()
-    })?;
-    execute_query_sync(client, sql)
-}
-
-pub fn execute_query_sync(client: &DmClient, sql: &str) -> Result<QueryResult, String> {
-    let start = Instant::now();
-    let sql = sql.trim().trim_end_matches(';');
-
-    if starts_with_executable_sql_keyword(sql, &["SELECT", "WITH", "SHOW", "DESCRIBE", "EXPLAIN"]) {
-        match client.conn.execute(sql, (), None).map_err(|e| e.to_string())? {
-            Some(mut cursor) => {
-                let col_count = cursor.num_result_cols().map_err(|e| e.to_string())? as u16;
-                let columns: Vec<String> = (1..=col_count)
-                    .map(|i| cursor.col_name(i).map_err(|e| e.to_string()).unwrap_or_else(|_| format!("col{i}")))
-                    .collect();
-
-                let buffer = TextRowSet::for_cursor(1000, &mut cursor, Some(8192)).map_err(|e| e.to_string())?;
-                let mut row_cursor = cursor.bind_buffer(buffer).map_err(|e| e.to_string())?;
-
-                let mut rows = Vec::new();
-                while let Some(batch) = row_cursor.fetch().map_err(|e| e.to_string())? {
-                    for row_idx in 0..batch.num_rows() {
-                        let vals: Vec<serde_json::Value> = (0..col_count as usize)
-                            .map(|col| {
-                                batch
-                                    .at(col, row_idx)
-                                    .and_then(|bytes| std::str::from_utf8(bytes).ok())
-                                    .map(|s| serde_json::Value::String(s.to_string()))
-                                    .unwrap_or(serde_json::Value::Null)
-                            })
-                            .collect();
-                        rows.push(vals);
-                        if rows.len() >= crate::query::MAX_ROWS {
-                            break;
-                        }
-                    }
-                    if rows.len() >= crate::query::MAX_ROWS {
-                        break;
-                    }
-                }
-
-                let truncated = rows.len() >= crate::query::MAX_ROWS;
-                Ok(QueryResult {
-                    columns,
-                    rows,
-                    affected_rows: 0,
-                    execution_time_ms: start.elapsed().as_millis(),
-                    truncated,
-                })
-            }
-            None => Ok(QueryResult {
-                columns: vec![],
-                rows: vec![],
-                affected_rows: 0,
-                execution_time_ms: start.elapsed().as_millis(),
-                truncated: false,
-            }),
-        }
-    } else {
-        match client.conn.execute(sql, (), None) {
-            Ok(Some(_cursor)) => Ok(QueryResult {
-                columns: vec![],
-                rows: vec![],
-                affected_rows: 0,
-                execution_time_ms: start.elapsed().as_millis(),
-                truncated: false,
-            }),
-            Ok(None) => Ok(QueryResult {
-                columns: vec![],
-                rows: vec![],
-                affected_rows: 0,
-                execution_time_ms: start.elapsed().as_millis(),
-                truncated: false,
-            }),
-            Err(e) => Err(e.to_string()),
-        }
-    }
-}

crates/dbx-core/src/db/mod.rs

@@ -1,5 +1,5 @@
+pub mod agent_driver;
 pub mod clickhouse_driver;
-pub mod dm_driver;
 pub mod duckdb_driver;
 pub mod elasticsearch_driver;
 pub mod file_validator;

crates/dbx-core/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod agent_manager;
 pub mod ai;
 pub mod connection;
 pub mod connection_secrets;

crates/dbx-core/src/models/connection.rs

@@ -113,6 +113,9 @@ pub enum DatabaseType {
     StarRocks,
     Redshift,
     Dameng,
+    Kingbase,
+    Vastbase,
+    Goldendb,
     Gaussdb,
     Jdbc,
 }
@@ -142,6 +145,7 @@ impl ConnectionConfig {
             },
             DatabaseType::Redshift => Some("dev"),
             DatabaseType::Gaussdb => Some("postgres"),
+            DatabaseType::Kingbase | DatabaseType::Vastbase => Some("postgres"),
             _ => None,
         }
     }
@@ -208,6 +212,9 @@ impl ConnectionConfig {
             DatabaseType::Oracle => format!("oracle://{host}:{port}{db_part}"),
             DatabaseType::Elasticsearch => format!("http://{host}:{port}"),
             DatabaseType::Dameng => format!("dm://{host}:{port}{db_part}"),
+            DatabaseType::Kingbase => format!("kingbase://{host}:{port}{db_part}"),
+            DatabaseType::Vastbase => format!("vastbase://{host}:{port}{db_part}"),
+            DatabaseType::Goldendb => format!("goldendb://{host}:{port}{db_part}"),
             DatabaseType::Gaussdb => format!("gaussdb://{host}:{port}{db_part}"),
             DatabaseType::Jdbc => "jdbc:<redacted>".to_string(),
         }
@@ -277,6 +284,15 @@ impl ConnectionConfig {
             DatabaseType::Dameng => {
                 format!("dm://{}:{}@{host}:{port}{db_part}", username, password)
             }
+            DatabaseType::Kingbase => {
+                format!("kingbase://{}:{}@{host}:{port}{db_part}", username, password)
+            }
+            DatabaseType::Vastbase => {
+                format!("vastbase://{}:{}@{host}:{port}{db_part}", username, password)
+            }
+            DatabaseType::Goldendb => {
+                format!("goldendb://{}:{}@{host}:{port}{db_part}", username, password)
+            }
             DatabaseType::Gaussdb => {
                 format!("gaussdb://{}:{}@{host}:{port}{db_part}", username, password)
             }

crates/dbx-core/src/query.rs

@@ -233,21 +233,18 @@ pub async fn do_execute(
         }
         PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()),
         PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()),
-        PoolKind::Dameng(client) => {
+        PoolKind::Agent(client) => {
             let client = client.clone();
             let sql = sql.to_string();
             let schema = schema.map(|s| s.to_string());
             drop(connections);
             wait_for_query(cancel_token, async move {
-                let task = tokio::task::spawn_blocking(move || {
-                    let client = client.lock().map_err(|e| e.to_string())?;
-                    if let Some(schema) = schema {
-                        db::dm_driver::execute_query_with_schema_sync(&client, &schema, &sql)
-                    } else {
-                        db::dm_driver::execute_query_sync(&client, &sql)
-                    }
-                });
-                task.await.map_err(|e| e.to_string())?
+                let mut client = client.lock().await;
+                let params = match schema {
+                    Some(s) => serde_json::json!({"sql": sql, "schema": s}),
+                    None => serde_json::json!({"sql": sql}),
+                };
+                client.call("execute_query", params).await
             })
             .await
             .map(truncate_result)
@@ -501,7 +498,7 @@ pub async fn execute_statements(
 /// Execute multiple SQL statements within a single transaction.
 /// For sqlx-based pools (Postgres/MySQL/SQLite), uses the Transaction API to
 /// guarantee all statements run on the same physical connection.
-/// For custom drivers (ClickHouse/SqlServer/Dameng/Gaussdb), uses explicit
+/// For custom drivers (ClickHouse/SqlServer/Agent/Gaussdb), uses explicit
 /// BEGIN/COMMIT/ROLLBACK on the already-single-connection client.
 /// For databases that don't support explicit transactions (Redis, MongoDB, Oracle),
 /// executes statements sequentially without transaction.
@@ -528,7 +525,7 @@ pub async fn execute_statements_in_transaction(
             PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()),
             PoolKind::Mysql(mp, _mode) => TxPath::Mysql(mp.clone(), false),
             PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()),
-            PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Dameng(_) | PoolKind::Gaussdb(_) => {
+            PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Agent(_) | PoolKind::Gaussdb(_) => {
                 TxPath::Explicit
             }
             PoolKind::DuckDb(_)

crates/dbx-core/src/schema.rs

@@ -109,12 +109,12 @@ pub fn extract_oracle(connections: &HashMap<String, PoolKind>, key: &str) -> Opt
     }
 }
 
-pub fn extract_dameng(
+pub fn extract_agent(
     connections: &HashMap<String, PoolKind>,
     key: &str,
-) -> Option<Arc<std::sync::Mutex<db::dm_driver::DmClient>>> {
+) -> Option<Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>> {
     match connections.get(key)? {
-        PoolKind::Dameng(client) => Some(client.clone()),
+        PoolKind::Agent(client) => Some(client.clone()),
         _ => None,
     }
 }
@@ -158,10 +158,10 @@ pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Resul
             let client = client.lock().await;
             return db::oracle_driver::list_databases(&*client).await;
         }
-        if let Some(client) = extract_dameng(&connections, connection_id) {
+        if let Some(client) = extract_agent(&connections, connection_id) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_databases(&client);
+            let mut client = client.lock().await;
+            return client.call("list_databases", serde_json::json!({})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, connection_id) {
             drop(connections);
@@ -212,10 +212,10 @@ pub async fn list_schemas_core(state: &AppState, connection_id: &str, database:
             let client = client.lock().await;
             return db::oracle_driver::list_schemas(&*client).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_schemas(&client);
+            let mut client = client.lock().await;
+            return client.call("list_schemas", serde_json::json!({"database": database})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -284,10 +284,10 @@ pub async fn list_tables_core(
             let client = client.lock().await;
             return db::oracle_driver::list_tables(&*client, schema).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_tables(&client, schema);
+            let mut client = client.lock().await;
+            return client.call("list_tables", serde_json::json!({"schema": schema})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -424,10 +424,10 @@ pub async fn get_columns_core(
             let client = client.lock().await;
             return db::oracle_driver::get_columns(&*client, schema, table).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::get_columns(&client, schema, table);
+            let mut client = client.lock().await;
+            return client.call("get_columns", serde_json::json!({"schema": schema, "table": table})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -475,10 +475,10 @@ pub async fn list_indexes_core(
             let client = client.lock().await;
             return db::oracle_driver::list_indexes(&*client, schema, table).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_indexes(&client, schema, table);
+            let mut client = client.lock().await;
+            return client.call("list_indexes", serde_json::json!({"schema": schema, "table": table})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -526,10 +526,10 @@ pub async fn list_foreign_keys_core(
             let client = client.lock().await;
             return db::oracle_driver::list_foreign_keys(&*client, schema, table).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_foreign_keys(&client, schema, table);
+            let mut client = client.lock().await;
+            return client.call("list_foreign_keys", serde_json::json!({"schema": schema, "table": table})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -577,10 +577,10 @@ pub async fn list_triggers_core(
             let client = client.lock().await;
             return db::oracle_driver::list_triggers(&*client, schema, table).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return db::dm_driver::list_triggers(&client, schema, table);
+            let mut client = client.lock().await;
+            return client.call("list_triggers", serde_json::json!({"schema": schema, "table": table})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -654,10 +654,10 @@ pub async fn get_table_ddl_core(
             let client = client.lock().await;
             return build_oracle_ddl(&*client, schema, table).await;
         }
-        if let Some(client) = extract_dameng(&connections, &pool_key) {
+        if let Some(client) = extract_agent(&connections, &pool_key) {
             drop(connections);
-            let client = client.lock().map_err(|e| e.to_string())?;
-            return build_dameng_ddl(&client, schema, table);
+            let mut client = client.lock().await;
+            return client.call("get_table_ddl", serde_json::json!({"schema": schema, "table": table})).await;
         }
         if let Some(client) = extract_gaussdb(&connections, &pool_key) {
             drop(connections);
@@ -1099,53 +1099,6 @@ pub async fn build_oracle_ddl(
     Ok(ddl)
 }
 
-pub fn build_dameng_ddl(client: &db::dm_driver::DmClient, schema: &str, table: &str) -> Result<String, String> {
-    let columns = db::dm_driver::get_columns(client, schema, table)?;
-    let indexes = db::dm_driver::list_indexes(client, schema, table)?;
-    let fkeys = db::dm_driver::list_foreign_keys(client, schema, table)?;
-
-    let mut ddl = format!("CREATE TABLE \"{schema}\".\"{table}\" (\n");
-    let col_lines: Vec<String> = columns
-        .iter()
-        .map(|c| {
-            let mut line = format!("  \"{}\" {}", c.name, c.data_type);
-            if !c.is_nullable {
-                line.push_str(" NOT NULL");
-            }
-            if let Some(ref def) = c.column_default {
-                line.push_str(&format!(" DEFAULT {def}"));
-            }
-            line
-        })
-        .collect();
-    ddl.push_str(&col_lines.join(",\n"));
-
-    let pks: Vec<&str> = columns.iter().filter(|c| c.is_primary_key).map(|c| c.name.as_str()).collect();
-    if !pks.is_empty() {
-        ddl.push_str(&format!(
-            ",\n  PRIMARY KEY ({})",
-            pks.iter().map(|k| format!("\"{k}\"")).collect::<Vec<_>>().join(", ")
-        ));
-    }
-    for fk in &fkeys {
-        ddl.push_str(&format!(
-            ",\n  CONSTRAINT \"{}\" FOREIGN KEY (\"{}\") REFERENCES \"{}\"(\"{}\")",
-            fk.name, fk.column, fk.ref_table, fk.ref_column
-        ));
-    }
-    ddl.push_str("\n);\n");
-
-    for idx in &indexes {
-        if idx.is_primary {
-            continue;
-        }
-        let unique = if idx.is_unique { "UNIQUE " } else { "" };
-        let cols = idx.columns.iter().map(|c| format!("\"{c}\"")).collect::<Vec<_>>().join(", ");
-        ddl.push_str(&format!("\nCREATE {unique}INDEX \"{}\" ON \"{schema}\".\"{table}\" ({cols});", idx.name));
-    }
-    Ok(ddl)
-}
-
 pub async fn build_gaussdb_ddl(
     client: &mut db::gaussdb_driver::GaussdbClient,
     schema: &str,