feat(dbx-cli): 增加 runtime discovery 与命令分发

- 增加 dbx-cli 参数解析与 8 个命令入口

- 实现 runtime discovery client 与 headless fallback

- 更新 CLI 依赖和锁文件

- 纳入 cli/storage 的格式化 diff

crates/dbx-cli/Cargo.toml

@@ -15,3 +15,6 @@ serde_json = "1.0"
 tokio = { version = "1", features = ["full"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 uuid = { version = "1", features = ["v4", "serde"] }
+
+[dev-dependencies]
+tempfile = "3"

crates/dbx-cli/src/commands.rs

@@ -1,26 +1,219 @@
 use dbx_core::cli::{fail, ok, CliEnvelope, CliErrorCode, CliSource};
 
 pub(crate) async fn run(args: Vec<String>) -> Result<(), CliEnvelope<()>> {
-    let output = match args.as_slice() {
-        [cmd, ..] if cmd == "context" => context().await,
-        [cmd, ..] if cmd == "selection" => runtime_required("dbx selection requires DBX GUI runtime."),
-        [cmd, sub, ..] if cmd == "result" && sub == "current" => {
-            runtime_required("dbx result current requires DBX GUI runtime.")
-        }
-        _ => ok(CliSource::Headless, serde_json::json!({ "runtime": "headless" })),
-    };
-
+    let output = dispatch(args).await;
     println!("{}", serde_json::to_string_pretty(&output).unwrap());
+
+    if matches!(output, CliEnvelope::Failure { .. }) {
+        std::process::exit(1);
+    }
+
     Ok(())
 }
 
-async fn context() -> CliEnvelope<serde_json::Value> {
+pub(crate) async fn dispatch(args: Vec<String>) -> CliEnvelope<serde_json::Value> {
+    let parsed = match parse_args(args) {
+        Ok(parsed) => parsed,
+        Err(err) => return err,
+    };
+
+    match parsed.positionals.as_slice() {
+        [cmd, rest @ ..] if cmd == "context" => context(rest).await,
+        [cmd, sub, rest @ ..] if cmd == "conn" && sub == "list" => conn_list(rest).await,
+        [cmd, sub, name, rest @ ..] if cmd == "conn" && sub == "show" => conn_show(name, rest).await,
+        [cmd, sub, rest @ ..] if cmd == "schema" && sub == "snapshot" => schema_snapshot(rest).await,
+        [cmd, rest @ ..] if cmd == "safe-query" => safe_query(rest).await,
+        [cmd, rest @ ..] if cmd == "handoff" => handoff(rest).await,
+        [cmd, rest @ ..] if cmd == "selection" => selection(rest).await,
+        [cmd, sub, rest @ ..] if cmd == "result" && sub == "current" => result_current(rest).await,
+        _ => fail(CliSource::Headless, CliErrorCode::InternalError, "Unknown command", false),
+    }
+}
+
+struct ParsedArgs {
+    positionals: Vec<String>,
+}
+
+fn parse_args(args: Vec<String>) -> Result<ParsedArgs, CliEnvelope<serde_json::Value>> {
+    let mut positionals = Vec::new();
+    let mut index = 0;
+
+    while index < args.len() {
+        if args[index] == "--format" {
+            let Some(format) = args.get(index + 1) else {
+                return Err(fail(CliSource::Headless, CliErrorCode::InternalError, "--format requires a value", true));
+            };
+            if format != "json" {
+                return Err(fail(CliSource::Headless, CliErrorCode::InternalError, "Only --format json is supported", true));
+            }
+            index += 2;
+        } else {
+            positionals.push(args[index].clone());
+            index += 1;
+        }
+    }
+
+    Ok(ParsedArgs { positionals })
+}
+
+async fn context(_args: &[String]) -> CliEnvelope<serde_json::Value> {
     match crate::runtime_client::get_json("/context").await {
         Ok(data) => ok(CliSource::GuiRuntime, data),
         Err(_) => ok(CliSource::Headless, serde_json::json!({ "runtime": "headless" })),
     }
 }
 
+async fn conn_list(_args: &[String]) -> CliEnvelope<serde_json::Value> {
+    ok(CliSource::Headless, serde_json::json!({ "connections": [] }))
+}
+
+async fn conn_show(_name: &str, _args: &[String]) -> CliEnvelope<serde_json::Value> {
+    ok(CliSource::Headless, serde_json::json!({}))
+}
+
+async fn schema_snapshot(args: &[String]) -> CliEnvelope<serde_json::Value> {
+    let Some(_conn) = option_value(args, "--conn") else {
+        return fail(CliSource::Headless, CliErrorCode::ConnectionNotFound, "--conn is required", true);
+    };
+
+    fail(CliSource::Headless, CliErrorCode::InternalError, "Schema snapshot headless execution is not implemented", true)
+}
+
+async fn safe_query(args: &[String]) -> CliEnvelope<serde_json::Value> {
+    let Some(_conn) = option_value(args, "--conn") else {
+        return fail(CliSource::Headless, CliErrorCode::ConnectionNotFound, "--conn is required", true);
+    };
+    let Some(_sql) = option_value(args, "--sql") else {
+        return fail(CliSource::Headless, CliErrorCode::QueryClassificationFailed, "--sql is required", true);
+    };
+
+    fail(CliSource::Headless, CliErrorCode::InternalError, "Safe query headless execution is not implemented", true)
+}
+
+async fn handoff(args: &[String]) -> CliEnvelope<serde_json::Value> {
+    let body = serde_json::json!({
+        "conn": option_value(args, "--conn"),
+        "title": option_value(args, "--title"),
+        "sql": option_value(args, "--sql"),
+    });
+
+    match crate::runtime_client::post_json("/handoff", body).await {
+        Ok(data) => ok(CliSource::GuiRuntime, data),
+        Err(_) => fail(CliSource::Headless, CliErrorCode::InternalError, "Handoff payload is required", true),
+    }
+}
+
+async fn selection(_args: &[String]) -> CliEnvelope<serde_json::Value> {
+    match crate::runtime_client::get_json("/selection").await {
+        Ok(data) => ok(CliSource::GuiRuntime, data),
+        Err(_) => runtime_required("dbx selection requires DBX GUI runtime."),
+    }
+}
+
+async fn result_current(args: &[String]) -> CliEnvelope<serde_json::Value> {
+    let limit = option_value(args, "--limit").unwrap_or("50");
+
+    match crate::runtime_client::get_json(&format!("/result/current?limit={limit}")).await {
+        Ok(data) => ok(CliSource::GuiRuntime, data),
+        Err(_) => runtime_required("dbx result current requires DBX GUI runtime."),
+    }
+}
+
 fn runtime_required(message: &str) -> CliEnvelope<serde_json::Value> {
     fail(CliSource::Headless, CliErrorCode::GuiRuntimeRequired, message, true)
 }
+
+fn option_value<'a>(args: &'a [String], key: &str) -> Option<&'a str> {
+    args.windows(2)
+        .find(|pair| pair[0] == key)
+        .map(|pair| pair[1].as_str())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use dbx_core::cli::CliErrorCode;
+    use std::sync::Mutex;
+
+    static ENV_LOCK: Mutex<()> = Mutex::new(());
+
+    fn assert_failure_code(env: CliEnvelope<serde_json::Value>, expected: CliErrorCode) {
+        match env {
+            CliEnvelope::Failure { error, .. } => assert_eq!(error.code, expected),
+            CliEnvelope::Success { .. } => panic!("expected failure envelope"),
+        }
+    }
+
+    #[tokio::test]
+    async fn gui_only_commands_return_runtime_required_without_runtime() {
+        let _guard = ENV_LOCK.lock().unwrap();
+        let dir = tempfile::tempdir().unwrap();
+        std::env::set_var("DBX_APP_DATA_DIR", dir.path());
+
+        assert_failure_code(
+            dispatch(vec!["selection".into(), "--format".into(), "json".into()]).await,
+            CliErrorCode::GuiRuntimeRequired,
+        );
+        assert_failure_code(
+            dispatch(vec![
+                "result".into(),
+                "current".into(),
+                "--limit".into(),
+                "25".into(),
+                "--format".into(),
+                "json".into(),
+            ])
+            .await,
+            CliErrorCode::GuiRuntimeRequired,
+        );
+
+        std::env::remove_var("DBX_APP_DATA_DIR");
+    }
+
+    #[tokio::test]
+    async fn recognizes_all_eight_cli_commands_with_json_format() {
+        let _guard = ENV_LOCK.lock().unwrap();
+        let dir = tempfile::tempdir().unwrap();
+        std::env::set_var("DBX_APP_DATA_DIR", dir.path());
+
+        let cases = [
+            vec!["context", "--format", "json"],
+            vec!["conn", "list", "--format", "json"],
+            vec!["conn", "show", "__missing__", "--redacted", "--format", "json"],
+            vec!["schema", "snapshot", "--format", "json"],
+            vec!["safe-query", "--format", "json"],
+            vec!["handoff", "--format", "json"],
+            vec!["selection", "--format", "json"],
+            vec!["result", "current", "--limit", "50", "--format", "json"],
+        ];
+
+        for args in cases {
+            let env = dispatch(args.iter().map(|value| value.to_string()).collect()).await;
+            let json = serde_json::to_value(&env).unwrap();
+            assert!(json.get("ok").is_some(), "missing ok for args: {args:?}");
+            assert!(json.get("source").is_some(), "missing source for args: {args:?}");
+            assert!(
+                json.get("data").is_some() || json.get("error").is_some(),
+                "missing data/error for args: {args:?}"
+            );
+        }
+
+        std::env::remove_var("DBX_APP_DATA_DIR");
+    }
+
+    #[tokio::test]
+    async fn unknown_command_returns_internal_error_envelope() {
+        assert_failure_code(
+            dispatch(vec!["not-a-command".into(), "--format".into(), "json".into()]).await,
+            CliErrorCode::InternalError,
+        );
+    }
+
+    #[tokio::test]
+    async fn rejects_non_json_format() {
+        assert_failure_code(
+            dispatch(vec!["context".into(), "--format".into(), "text".into()]).await,
+            CliErrorCode::InternalError,
+        );
+    }
+}

crates/dbx-cli/src/runtime_client.rs

@@ -9,6 +9,10 @@ pub struct RuntimeDiscovery {
 }
 
 pub fn app_data_dir() -> PathBuf {
+    if let Ok(path) = std::env::var("DBX_APP_DATA_DIR") {
+        return PathBuf::from(path);
+    }
+
     let home = std::env::var(if cfg!(windows) { "APPDATA" } else { "HOME" })
         .unwrap_or_else(|_| ".".to_string());
 
@@ -31,12 +35,42 @@ pub async fn get_json(path: &str) -> Result<serde_json::Value, String> {
     let runtime = load_runtime().ok_or_else(|| "runtime unavailable".to_string())?;
     let url = format!("http://127.0.0.1:{}{}", runtime.port, path);
 
-    reqwest::Client::new()
+    let response = reqwest::Client::new()
         .get(url)
         .bearer_auth(runtime.token)
         .send()
         .await
-        .map_err(|err| err.to_string())?
+        .map_err(|err| err.to_string())?;
+
+    let status = response.status();
+    if !status.is_success() {
+        return Err(format!("runtime request failed with status {status}"));
+    }
+
+    response
+        .json()
+        .await
+        .map_err(|err| err.to_string())
+}
+
+pub async fn post_json(path: &str, body: serde_json::Value) -> Result<serde_json::Value, String> {
+    let runtime = load_runtime().ok_or_else(|| "runtime unavailable".to_string())?;
+    let url = format!("http://127.0.0.1:{}{}", runtime.port, path);
+
+    let response = reqwest::Client::new()
+        .post(url)
+        .bearer_auth(runtime.token)
+        .json(&body)
+        .send()
+        .await
+        .map_err(|err| err.to_string())?;
+
+    let status = response.status();
+    if !status.is_success() {
+        return Err(format!("runtime request failed with status {status}"));
+    }
+
+    response
         .json()
         .await
         .map_err(|err| err.to_string())

crates/dbx-core/src/cli.rs

@@ -116,12 +116,8 @@ where
         let source = source.ok_or_else(|| de::Error::missing_field("source"))?;
 
         match (ok, has_data, has_error) {
-            (true, true, false) => {
-                Ok(CliEnvelope::Success { ok, source, data: data.expect("data presence was checked") })
-            }
-            (false, false, true) => {
-                Ok(CliEnvelope::Failure { ok, source, error: error.expect("error presence was checked") })
-            }
+            (true, true, false) => Ok(CliEnvelope::Success { ok, source, data: data.expect("data presence was checked") }),
+            (false, false, true) => Ok(CliEnvelope::Failure { ok, source, error: error.expect("error presence was checked") }),
             (true, _, _) => Err(de::Error::custom("ok=true requires data without error")),
             (false, _, _) => Err(de::Error::custom("ok=false requires error without data")),
         }

crates/dbx-core/src/storage.rs

@@ -144,11 +144,10 @@ async fn ensure_history_columns(pool: &SqlitePool) -> Result<(), String> {
 }
 
 async fn ensure_handoffs_sequence(pool: &SqlitePool) -> Result<(), String> {
-    let (seq_columns,): (i64,) =
-        sqlx::query_as("SELECT COUNT(*) FROM pragma_table_info('handoffs') WHERE name = 'seq'")
-            .fetch_one(pool)
-            .await
-            .map_err(|e| e.to_string())?;
+    let (seq_columns,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM pragma_table_info('handoffs') WHERE name = 'seq'")
+        .fetch_one(pool)
+        .await
+        .map_err(|e| e.to_string())?;
 
     if seq_columns > 0 {
         return Ok(());
@@ -1058,8 +1057,10 @@ mod handoff_tests {
     async fn handoffs_table_has_stable_autoincrement_sequence() {
         let storage = open_temp_storage().await;
 
-        let columns: Vec<(String,)> =
-            sqlx::query_as("SELECT name FROM pragma_table_info('handoffs')").fetch_all(&storage.db).await.unwrap();
+        let columns: Vec<(String,)> = sqlx::query_as("SELECT name FROM pragma_table_info('handoffs')")
+            .fetch_all(&storage.db)
+            .await
+            .unwrap();
 
         assert!(columns.iter().any(|(name,)| name == "seq"));
     }