feat(mongodb): support generic runCommand execution

crates/dbx-core/src/agent_tools.rs

@@ -1387,6 +1387,7 @@ for line in sys.stdin:
             "db.items.updateMany({tenant: 7}, {$set: {active: false}})",
             "db.items.deleteMany({tenant: 7})",
             "db.items.createIndex({tenant: 1})",
+            "db.runCommand({compact: 'items'})",
             r#"db.items.aggregate([{"$out":"items_backup"}])"#,
         ]
         .into_iter()

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

@@ -973,6 +973,7 @@ pub enum AgentCapability {
     EtcdAuth,
     MongoDropDatabase,
     MongoCloneCollection,
+    MongoRunCommand,
     MultiSession,
     StructuredErrorV1,
 }
@@ -1055,7 +1056,7 @@ fn parse_agent_rpc_error_header(header: &str) -> (Option<i64>, String) {
 }
 
 impl AgentCapability {
-    pub const ALL: [Self; 22] = [
+    pub const ALL: [Self; 23] = [
         Self::Connect,
         Self::TestConnection,
         Self::Metadata,
@@ -1076,6 +1077,7 @@ impl AgentCapability {
         Self::EtcdAuth,
         Self::MongoDropDatabase,
         Self::MongoCloneCollection,
+        Self::MongoRunCommand,
         Self::MultiSession,
         Self::StructuredErrorV1,
     ];
@@ -1102,6 +1104,7 @@ impl AgentCapability {
             Self::EtcdAuth => "etcd_auth",
             Self::MongoDropDatabase => "mongo_drop_database",
             Self::MongoCloneCollection => "mongo_clone_collection",
+            Self::MongoRunCommand => "mongo_run_command",
             Self::MultiSession => "multi_session",
             Self::StructuredErrorV1 => "structured_error_v1",
         }
@@ -1294,10 +1297,11 @@ pub enum MongoAgentMethod {
     UpdateDocuments,
     DeleteDocument,
     DeleteDocuments,
+    RunCommand,
 }
 
 impl MongoAgentMethod {
-    pub const ALL: [Self; 20] = [
+    pub const ALL: [Self; 21] = [
         Self::ListDatabases,
         Self::ListCollections,
         Self::FindDocuments,
@@ -1318,6 +1322,7 @@ impl MongoAgentMethod {
         Self::UpdateDocuments,
         Self::DeleteDocument,
         Self::DeleteDocuments,
+        Self::RunCommand,
     ];
 
     pub fn as_str(self) -> &'static str {
@@ -1342,6 +1347,7 @@ impl MongoAgentMethod {
             Self::UpdateDocuments => "update_documents",
             Self::DeleteDocument => "delete_document",
             Self::DeleteDocuments => "delete_documents",
+            Self::RunCommand => "run_command",
         }
     }
 }
@@ -2539,6 +2545,13 @@ impl AgentDriverClient {
         self.call_mongo_method(MongoAgentMethod::ServerVersion, mongo_database_params(database)).await
     }
 
+    pub async fn mongo_run_command<T: DeserializeOwned + Send + 'static>(
+        &mut self,
+        params: Value,
+    ) -> Result<T, String> {
+        self.call_mongo_method(MongoAgentMethod::RunCommand, params).await
+    }
+
     pub async fn mongo_create_index<T: DeserializeOwned + Send + 'static>(
         &mut self,
         params: Value,
@@ -2755,6 +2768,7 @@ pub fn agent_supports_capability(handshake: Option<&AgentHandshake>, capability:
             | AgentCapability::EtcdAuth
             | AgentCapability::MongoDropDatabase
             | AgentCapability::MongoCloneCollection
+            | AgentCapability::MongoRunCommand
     ) {
         return handshake.map(|value| value.supports(capability)).unwrap_or(false);
     }
@@ -4343,9 +4357,10 @@ for line in sys.stdin:
         assert_eq!(AgentCapability::EtcdAuth.as_str(), "etcd_auth");
         assert_eq!(AgentCapability::MongoDropDatabase.as_str(), "mongo_drop_database");
         assert_eq!(AgentCapability::MongoCloneCollection.as_str(), "mongo_clone_collection");
+        assert_eq!(AgentCapability::MongoRunCommand.as_str(), "mongo_run_command");
         assert_eq!(AgentCapability::MultiSession.as_str(), "multi_session");
         assert_eq!(AgentCapability::StructuredErrorV1.as_str(), "structured_error_v1");
-        assert_eq!(AgentCapability::ALL.len(), 22);
+        assert_eq!(AgentCapability::ALL.len(), 23);
     }
 
     #[test]
@@ -4702,6 +4717,8 @@ for line in sys.stdin:
         assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoDropDatabase));
         assert!(!agent_supports_capability(None, AgentCapability::MongoCloneCollection));
         assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoCloneCollection));
+        assert!(!agent_supports_capability(None, AgentCapability::MongoRunCommand));
+        assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoRunCommand));
 
         let mongo_handshake =
             AgentHandshake { capabilities: vec![AgentCapability::MongoDropDatabase.as_str().to_string()], ..handshake };
@@ -4712,6 +4729,12 @@ for line in sys.stdin:
             ..mongo_handshake
         };
         assert!(agent_supports_capability(Some(&mongo_clone_handshake), AgentCapability::MongoCloneCollection));
+
+        let mongo_run_command_handshake = AgentHandshake {
+            capabilities: vec![AgentCapability::MongoRunCommand.as_str().to_string()],
+            ..mongo_clone_handshake
+        };
+        assert!(agent_supports_capability(Some(&mongo_run_command_handshake), AgentCapability::MongoRunCommand));
     }
 
     #[test]

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

@@ -173,6 +173,25 @@ pub async fn server_version(client: &Client, database: &str) -> Result<String, S
     server_version_from_build_info(&result)
 }
 
+pub async fn run_command(client: &Client, database: &str, command_json: &str) -> Result<MongoDocumentResult, String> {
+    let value: serde_json::Value =
+        serde_json::from_str(command_json).map_err(|error| format!("Invalid MongoDB command JSON: {error}"))?;
+    let command = json_object_to_document_extended_json(&value)
+        .map_err(|error| format!("Invalid MongoDB command document: {error}"))?;
+    if command.is_empty() {
+        return Err("MongoDB runCommand requires a non-empty command document".to_string());
+    }
+    let result = client.database(database).run_command(command).await.map_err(|error| error.to_string())?;
+    let (document, extended_document) = document_json_views(result);
+    Ok(MongoDocumentResult {
+        documents: vec![document],
+        raw_documents: None,
+        extended_documents: Some(vec![extended_document]),
+        total: 1,
+        total_is_exact: true,
+    })
+}
+
 fn server_version_from_build_info(result: &Document) -> Result<String, String> {
     result.get_str("version").map(str::to_string).map_err(|e| format!("MongoDB server version not found: {e}"))
 }

crates/dbx-core/src/mongo_ops.rs

@@ -148,6 +148,35 @@ pub async fn mongo_server_version_core(
     }
 }
 
+pub async fn mongo_run_command_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    command_json: &str,
+) -> Result<MongoDocumentResult, String> {
+    ensure_document_pool(state, connection_id).await?;
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::MongoDb(client) => mongo_driver::run_command(client, database, command_json).await,
+        PoolKind::Agent(client) => {
+            let mut client = client.lock().await;
+            if !client.supports_capability(AgentCapability::MongoRunCommand) {
+                return Err(
+                    "MongoDB Legacy Agent does not support runCommand; upgrade or reinstall the MongoDB Legacy driver"
+                        .to_string(),
+                );
+            }
+            client
+                .mongo_run_command(serde_json::json!({
+                    "database": database,
+                    "command_json": command_json,
+                }))
+                .await
+        }
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}
+
 pub async fn mongo_collection_stats_core(
     state: &AppState,
     connection_id: &str,
@@ -821,6 +850,10 @@ pub async fn execute_mongo_command_core(
             .await
             .map(|version| scalar_query_result("version", Value::String(version))),
         MongoCommand::Use { database } => Ok(scalar_query_result("database", Value::String(database.clone()))),
+        MongoCommand::RunCommand { command_json } => {
+            let result = mongo_run_command_core(state, connection_id, database, command_json).await?;
+            Ok(mongo_documents_query_result(result.documents))
+        }
         MongoCommand::Find { collection, filter, projection, sort, collation, skip, limit } => {
             let limit = bounded_mongo_find_limit(*limit, max_rows);
             let result = mongo_find_documents_without_total_core(
@@ -1347,6 +1380,55 @@ for line in sys.stdin:
         assert!(!error.contains("Unknown method"), "{error}");
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn mongo_run_command_routes_legacy_connections_to_the_agent() {
+        let expected_result = serde_json::json!({
+            "documents": [{"ok": 1, "cursor": {"firstBatch": [{"_id": {"$oid": "507f1f77bcf86cd799439011"}}]}}],
+            "extended_documents": [{"ok": 1, "cursor": {"firstBatch": [{"_id": {"$oid": "507f1f77bcf86cd799439011"}}]}}],
+            "total": 1,
+        });
+        let (state, _directory) = legacy_mongo_state_with_capabilities(
+            "run_command",
+            serde_json::json!({
+                "database": "app",
+                "command_json": "{\"ping\":1}",
+            }),
+            expected_result,
+            &[AgentCapability::MongoRunCommand.as_str()],
+        )
+        .await;
+
+        let result = mongo_run_command_core(&state, "legacy", "app", "{\"ping\":1}").await.unwrap();
+
+        assert_eq!(result.total, 1);
+        assert_eq!(result.documents[0]["ok"], 1);
+        assert_eq!(
+            result.extended_documents.as_ref().unwrap()[0]["cursor"]["firstBatch"][0]["_id"],
+            serde_json::json!({"$oid": "507f1f77bcf86cd799439011"})
+        );
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn mongo_run_command_requires_an_explicit_legacy_agent_capability() {
+        let (state, _directory) = legacy_mongo_state_with_capabilities(
+            "run_command",
+            serde_json::json!({
+                "database": "app",
+                "command_json": "{\"ping\":1}",
+            }),
+            serde_json::json!({"documents": [{"ok": 1}], "total": 1}),
+            &[],
+        )
+        .await;
+
+        let error = mongo_run_command_core(&state, "legacy", "app", "{\"ping\":1}").await.unwrap_err();
+
+        assert!(error.contains("upgrade or reinstall"), "{error}");
+        assert!(!error.contains("Unknown method"), "{error}");
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn mongo_drop_collection_routes_legacy_connections_to_the_agent() {

crates/dbx-core/src/mongo_shell.rs

@@ -8,6 +8,11 @@ pub enum MongoCommand {
     Version,
     #[serde(rename = "use")]
     Use { database: String },
+    #[serde(rename = "runCommand")]
+    RunCommand {
+        #[serde(rename = "commandJson")]
+        command_json: String,
+    },
     #[serde(rename = "createUser")]
     CreateUser {
         #[serde(rename = "userJson")]
@@ -84,7 +89,8 @@ impl MongoCommand {
     pub fn is_mutating(&self) -> bool {
         matches!(
             self,
-            Self::CreateUser { .. }
+            Self::RunCommand { .. }
+                | Self::CreateUser { .. }
                 | Self::Insert { .. }
                 | Self::Update { .. }
                 | Self::Delete { .. }
@@ -98,7 +104,7 @@ impl MongoCommand {
     }
 
     pub fn is_dangerous(&self) -> bool {
-        matches!(self, Self::CreateUser { .. } | Self::DropCollection { .. })
+        matches!(self, Self::RunCommand { .. } | Self::CreateUser { .. } | Self::DropCollection { .. })
             || matches!(self, Self::DropIndexes { indexes: None, single: false, .. })
             || matches!(self, Self::Aggregate { pipeline, .. } if aggregate_writes(pipeline))
     }
@@ -356,6 +362,19 @@ pub fn parse(input: &str) -> Result<MongoCommand, String> {
     if let Some(database) = parse_use_database(source) {
         return Ok(MongoCommand::Use { database });
     }
+    if let Some((args, tail)) = database_method_call(source, "runCommand") {
+        if !tail.is_empty() || args.len() != 1 {
+            return Err("MongoDB runCommand() requires exactly one command document.".to_string());
+        }
+        let command_json = normalized_json(&args[0])?;
+        let command = parse_json_value(&command_json)
+            .and_then(|value| value.as_object().cloned())
+            .ok_or("MongoDB runCommand() requires a command document.")?;
+        if command.is_empty() {
+            return Err("MongoDB runCommand() requires a non-empty command document.".to_string());
+        }
+        return Ok(MongoCommand::RunCommand { command_json });
+    }
     if let Some((args, tail)) = database_method_call(source, "createUser") {
         if !tail.is_empty() || !(1..=2).contains(&args.len()) {
             return Err("MongoDB createUser() requires a user document and optional write concern.".to_string());
@@ -1022,6 +1041,43 @@ mod tests {
         assert!(parse(r#"db.createUser({user: "test"}, "majority")"#).is_err());
     }
 
+    #[test]
+    fn parses_run_command_as_a_dangerous_write() {
+        let command = parse(
+            r#"db.runCommand({
+                find: "orders",
+                filter: {_id: ObjectId("507f1f77bcf86cd799439011")},
+                createdAt: ISODate("2025-01-01T00:00:00Z")
+            })"#,
+        )
+        .unwrap();
+        assert_eq!(
+            command,
+            MongoCommand::RunCommand {
+                command_json: r#"{"find":"orders","filter":{"_id":{"$oid":"507f1f77bcf86cd799439011"}},"createdAt":{"$date":"2025-01-01T00:00:00Z"}}"#.to_string(),
+            }
+        );
+        assert!(command.is_mutating());
+        assert!(command.is_dangerous());
+        assert_eq!(validate_safety(&command, false, false, false), Err(MongoSafetyError::WritesDisabled));
+        assert_eq!(validate_safety(&command, true, false, false), Err(MongoSafetyError::Dangerous));
+        assert_eq!(validate_safety(&command, true, true, false), Ok(()));
+    }
+
+    #[test]
+    fn rejects_unsupported_run_command_shapes() {
+        for source in [
+            "db.runCommand()",
+            "db.runCommand({})",
+            "db.runCommand('ping')",
+            "db.runCommand({ping: 1}, {readPreference: 'primary'})",
+            "db.runCommand({ping: 1}).valueOf()",
+            "db.runCommand([1, 2, 3])",
+        ] {
+            assert!(parse(source).unwrap_err().contains("runCommand"), "{source}");
+        }
+    }
+
     #[test]
     fn treats_effectively_unbounded_write_filters_as_dangerous() {
         for command in [
@@ -1171,6 +1227,9 @@ mod tests {
             serde_json::to_value(parse(r#"db.createUser({user: "app", pwd: "secret", roles: []})"#).unwrap()).unwrap();
         assert_eq!(create_user["kind"], "createUser");
         assert_eq!(create_user["userJson"], r#"{"user":"app","pwd":"secret","roles":[]}"#);
+        let run_command = serde_json::to_value(parse("db.runCommand({ping: 1})").unwrap()).unwrap();
+        assert_eq!(run_command["kind"], "runCommand");
+        assert_eq!(run_command["commandJson"], r#"{"ping":1}"#);
     }
 
     #[test]