feat(rabbitmq): add message queue management support

crates/dbx-core/assets/database-drivers.manifest.json

@@ -1721,6 +1721,11 @@
           "profile": "rocketmq",
           "label": "Apache RocketMQ",
           "agentKey": "rocketmq"
+        },
+        {
+          "profile": "rabbitmq",
+          "label": "RabbitMQ",
+          "agentKey": "rabbitmq"
         }
       ],
       "supportLevel": "connect",

crates/dbx-core/src/agent_catalog.rs

@@ -42,11 +42,13 @@ const H2_PROFILES: &[AgentDriverProfile] =
 const EXTRA_AGENT_LABELS: &[(&str, &str)] = &[
     ("kafka", "Apache Kafka"),
     ("rocketmq", "Apache RocketMQ"),
+    ("rabbitmq", "RabbitMQ"),
     ("sqlserver-legacy", "SQL Server legacy compatibility component"),
 ];
 const EXTRA_DRIVER_STORE_ENTRIES: &[(&str, &str)] = &[
     ("kafka", "Apache Kafka"),
     ("rocketmq", "Apache RocketMQ"),
+    ("rabbitmq", "RabbitMQ"),
     ("sqlserver-legacy", "SQL Server legacy compatibility component"),
 ];
 
@@ -310,6 +312,7 @@ pub fn agent_key(db_type: &DatabaseType, driver_profile: Option<&str>) -> Option
         return match driver_profile {
             Some("kafka") => Some("kafka"),
             Some("rocketmq") => Some("rocketmq"),
+            Some("rabbitmq") => Some("rabbitmq"),
             _ => None,
         };
     }

crates/dbx-core/src/mq/README.md

@@ -14,6 +14,7 @@ mq/
 ├── service.rs          - 服务层函数
 └── adapters/
     ├── pulsar.rs       - Pulsar 实现
+    ├── rabbitmq.rs     - RabbitMQ 实现 (Java agent)
     └── pulsar_version.rs - 版本探测
 ```
 

crates/dbx-core/src/mq/adapters/kafka.rs

@@ -45,6 +45,11 @@ const KAFKA_CAPABILITIES: MqCapabilities = MqCapabilities {
     supports_message_query: false,
     supports_dlq: false,
     supports_message_trace: false,
+    supports_exchanges: false,
+    supports_client_connections: false,
+    supports_user_permissions: false,
+    supports_policies: false,
+    supports_cluster_monitoring: false,
 };
 
 pub struct KafkaAdmin {
@@ -185,6 +190,7 @@ impl MessageQueueAdmin for KafkaAdmin {
                     persistent: true,
                     internal: t.get("internal").and_then(|v| v.as_bool()).unwrap_or(false),
                     message_type: None,
+                    namespace: None,
                 }
             })
             .collect())

crates/dbx-core/src/mq/adapters/mod.rs

@@ -3,4 +3,5 @@
 pub mod kafka;
 pub mod pulsar;
 pub mod pulsar_version;
+pub mod rabbitmq;
 pub mod rocketmq;

crates/dbx-core/src/mq/adapters/pulsar.rs

@@ -293,6 +293,7 @@ impl MessageQueueAdmin for PulsarAdmin {
                         persistent,
                         internal: false,
                         message_type: None,
+                        namespace: None,
                     })
                 })
                 .buffered(PARTITION_METADATA_CONCURRENCY)
@@ -318,6 +319,7 @@ impl MessageQueueAdmin for PulsarAdmin {
                     persistent,
                     internal: false,
                     message_type: None,
+                    namespace: None,
                 });
             }
         }

crates/dbx-core/src/mq/adapters/pulsar_version.rs

@@ -143,6 +143,11 @@ impl PulsarApiProfile {
                 supports_message_query: false,
                 supports_dlq: false,
                 supports_message_trace: false,
+                supports_exchanges: false,
+                supports_client_connections: false,
+                supports_user_permissions: false,
+                supports_policies: false,
+                supports_cluster_monitoring: false,
             },
         }
     }

crates/dbx-core/src/mq/adapters/rocketmq.rs

@@ -45,6 +45,11 @@ const ROCKETMQ_CAPABILITIES: MqCapabilities = MqCapabilities {
     supports_message_query: true,
     supports_dlq: true,
     supports_message_trace: true,
+    supports_exchanges: false,
+    supports_client_connections: false,
+    supports_user_permissions: false,
+    supports_policies: false,
+    supports_cluster_monitoring: false,
 };
 
 const TOPIC_LIST_PAGE_SIZE: u32 = 200;
@@ -726,6 +731,7 @@ fn topic_info_from_agent_value(t: &serde_json::Value) -> TopicInfo {
         persistent: true,
         internal: t.get("internal").and_then(|v| v.as_bool()).unwrap_or(false),
         message_type: t.get("messageType").and_then(|v| v.as_str()).map(String::from),
+        namespace: None,
     }
 }
 

crates/dbx-core/src/mq/config.rs

@@ -63,10 +63,11 @@ impl MqAdminConfig {
         let mut parsed: MqAdminConfig = serde_json::from_value(raw.clone())
             .map_err(|e| format!("Failed to parse message queue admin config: {e}"))?;
         parsed.admin_url = parsed.admin_url.trim().to_string();
-        // Kafka and RocketMQ use namesrv/bootstrap from `extra` instead of an admin URL.
+        // Kafka, RocketMQ and RabbitMQ use namesrv/bootstrap/addresses from `extra` instead of an admin URL.
         if parsed.admin_url.is_empty()
             && parsed.system_kind != MqSystemKind::Kafka
             && parsed.system_kind != MqSystemKind::RocketMq
+            && parsed.system_kind != MqSystemKind::RabbitMq
         {
             return Err("Message queue admin URL is empty".to_string());
         }
@@ -216,6 +217,45 @@ mod tests {
         assert_eq!(mqc.extra.get("namesrvAddr").and_then(|v| v.as_str()), Some("127.0.0.1:9876"));
     }
 
+    #[test]
+    fn parses_rabbitmq_config_with_empty_admin_url() {
+        let cfg = connection_with_external(serde_json::json!({
+            "systemKind": "rabbitmq",
+            "adminUrl": "",
+            "auth": { "kind": "basic", "username": "guest", "password": "guest" },
+            "extra": {
+                "addresses": "127.0.0.1",
+                "port": 5672,
+                "virtualHost": "/"
+            }
+        }));
+        let mqc = MqAdminConfig::from_connection(&cfg).expect("should parse valid RabbitMQ config");
+        assert_eq!(mqc.system_kind, MqSystemKind::RabbitMq);
+        assert_eq!(mqc.admin_url, "");
+        assert_eq!(mqc.extra.get("addresses").and_then(|v| v.as_str()), Some("127.0.0.1"));
+        assert_eq!(mqc.extra.get("virtualHost").and_then(|v| v.as_str()), Some("/"));
+    }
+
+    #[test]
+    fn parses_rabbitmq_config_with_management_admin_url() {
+        // An explicit management URL stays untouched: it may carry a reverse
+        // proxy path prefix, and no http(s)-only scheme restriction applies.
+        let cfg = connection_with_external(serde_json::json!({
+            "systemKind": "rabbitmq",
+            "adminUrl": "http://rabbit.internal:15672/proxy",
+            "auth": { "kind": "basic", "username": "guest", "password": "guest" },
+            "extra": {
+                "addresses": "127.0.0.1",
+                "port": 5672,
+                "virtualHost": "/"
+            }
+        }));
+        let mqc = MqAdminConfig::from_connection(&cfg).expect("should parse RabbitMQ config with a management URL");
+        assert_eq!(mqc.system_kind, MqSystemKind::RabbitMq);
+        assert_eq!(mqc.admin_url, "http://rabbit.internal:15672/proxy");
+        assert_eq!(mqc.extra.get("addresses").and_then(|v| v.as_str()), Some("127.0.0.1"));
+    }
+
     #[test]
     fn admin_url_with_endpoint_preserves_scheme_path_and_query() {
         let rewritten =

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

@@ -29,6 +29,7 @@ use crate::db::agent_driver::AgentLaunchSpec;
 use crate::models::connection::ConnectionConfig;
 use crate::mq::adapters::kafka::KafkaAdmin;
 use crate::mq::adapters::pulsar::PulsarAdmin;
+use crate::mq::adapters::rabbitmq::RabbitMqAdmin;
 use crate::mq::adapters::rocketmq::RocketMqAdmin;
 use crate::mq::config::MqAdminConfig;
 use crate::mq::port::MessageQueueAdmin;
@@ -156,5 +157,12 @@ async fn build_adapter(
             let adapter = RocketMqAdmin::new(mqc, launch).await?;
             Ok(Arc::new(adapter))
         }
+        MqSystemKindInternal::RabbitMq => {
+            let launch = agent_launch.ok_or(
+                "RabbitMQ adapter requires an agent launch spec. The RabbitMQ agent driver is not installed or not configured.",
+            )?;
+            let adapter = RabbitMqAdmin::new(mqc, launch).await?;
+            Ok(Arc::new(adapter))
+        }
     }
 }

crates/dbx-core/src/mq/port.rs

@@ -43,6 +43,62 @@ pub trait MessageQueueAdmin: Send + Sync {
         Err("Topic route is not supported by this MQ system".to_string())
     }
 
+    // ---- Exchanges / bindings (RabbitMQ) ----
+
+    async fn list_exchanges(&self, _ns: &NamespaceRef) -> Result<Vec<MqExchangeInfo>, String> {
+        Err("Exchanges are not supported by this MQ system".to_string())
+    }
+
+    async fn create_exchange(
+        &self,
+        _ns: &NamespaceRef,
+        _name: &str,
+        _exchange_type: &str,
+        _durable: bool,
+        _auto_delete: bool,
+    ) -> Result<(), String> {
+        Err("Exchanges are not supported by this MQ system".to_string())
+    }
+
+    async fn delete_exchange(&self, _ns: &NamespaceRef, _name: &str) -> Result<(), String> {
+        Err("Exchanges are not supported by this MQ system".to_string())
+    }
+
+    async fn list_bindings(
+        &self,
+        _ns: &NamespaceRef,
+        _exchange: Option<&str>,
+        _queue: Option<&str>,
+    ) -> Result<Vec<MqBindingInfo>, String> {
+        Err("Bindings are not supported by this MQ system".to_string())
+    }
+
+    async fn bind_queue(&self, _ns: &NamespaceRef, _binding: &MqBindingInfo) -> Result<(), String> {
+        Err("Bindings are not supported by this MQ system".to_string())
+    }
+
+    async fn unbind_queue(&self, _ns: &NamespaceRef, _binding: &MqBindingInfo) -> Result<(), String> {
+        Err("Bindings are not supported by this MQ system".to_string())
+    }
+
+    // ---- Client connections / channels (RabbitMQ) ----
+
+    async fn list_client_connections(&self, _ns: &NamespaceRef) -> Result<Vec<MqClientConnectionInfo>, String> {
+        Err("Client connections are not supported by this MQ system".to_string())
+    }
+
+    async fn list_client_channels(
+        &self,
+        _ns: &NamespaceRef,
+        _connection: Option<String>,
+    ) -> Result<Vec<MqChannelInfo>, String> {
+        Err("Client channels are not supported by this MQ system".to_string())
+    }
+
+    async fn close_client_connection(&self, _ns: &NamespaceRef, _name: &str) -> Result<(), String> {
+        Err("Closing client connections is not supported by this MQ system".to_string())
+    }
+
     async fn alter_topic_config(&self, _topic: &TopicRef, _configs: serde_json::Value) -> Result<(), String> {
         Err("Alter topic config is not supported by this MQ system".to_string())
     }
@@ -141,6 +197,65 @@ pub trait MessageQueueAdmin: Send + Sync {
     /// typed methods do not.
     async fn raw_request(&self, req: MqRawRequest) -> Result<MqRawResponse, String>;
 
+    // ---- Users & virtual-host permissions (RabbitMQ) ----
+
+    async fn list_users(&self) -> Result<Vec<MqUserInfo>, String> {
+        Err("User management is not supported by this MQ system".to_string())
+    }
+
+    async fn create_user(&self, _name: &str, _password: &str, _tags: Vec<String>) -> Result<(), String> {
+        Err("User management is not supported by this MQ system".to_string())
+    }
+
+    async fn delete_user(&self, _name: &str) -> Result<(), String> {
+        Err("User management is not supported by this MQ system".to_string())
+    }
+
+    async fn list_user_permissions(&self, _ns: &NamespaceRef) -> Result<Vec<MqVhostPermission>, String> {
+        Err("User permissions are not supported by this MQ system".to_string())
+    }
+
+    async fn grant_user_permission(
+        &self,
+        _ns: &NamespaceRef,
+        _user: &str,
+        _configure: &str,
+        _write: &str,
+        _read: &str,
+    ) -> Result<(), String> {
+        Err("User permissions are not supported by this MQ system".to_string())
+    }
+
+    async fn revoke_user_permission(&self, _ns: &NamespaceRef, _user: &str) -> Result<(), String> {
+        Err("User permissions are not supported by this MQ system".to_string())
+    }
+
+    // ---- Policies (RabbitMQ) ----
+
+    async fn list_policies(&self, _ns: &NamespaceRef) -> Result<Vec<MqPolicyInfo>, String> {
+        Err("Policies are not supported by this MQ system".to_string())
+    }
+
+    async fn set_policy(&self, _ns: &NamespaceRef, _policy: &MqPolicyInfo) -> Result<(), String> {
+        Err("Policies are not supported by this MQ system".to_string())
+    }
+
+    async fn delete_policy(&self, _ns: &NamespaceRef, _name: &str) -> Result<(), String> {
+        Err("Policies are not supported by this MQ system".to_string())
+    }
+
+    // ---- Cluster monitoring (RabbitMQ) ----
+
+    /// Broker-wide queue totals and message rates.
+    async fn get_overview(&self) -> Result<MqOverviewInfo, String> {
+        Err("Cluster overview is not supported by this MQ system".to_string())
+    }
+
+    /// Cluster node listing with resource usage.
+    async fn list_nodes(&self) -> Result<Vec<MqNodeInfo>, String> {
+        Err("Cluster node monitoring is not supported by this MQ system".to_string())
+    }
+
     // ---- Message production ----
 
     /// Produce a message to a topic. Adapters that do not support message

crates/dbx-core/src/mq/service.rs

@@ -174,6 +174,211 @@ pub async fn mq_get_topic_internal_stats_core(
     adapter.get_topic_internal_stats(&topic).await
 }
 
+// ---- Exchanges / bindings (RabbitMQ) ----
+
+pub async fn mq_list_exchanges_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+) -> Result<Vec<MqExchangeInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_exchanges(&ns).await
+}
+
+pub async fn mq_create_exchange_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    name: &str,
+    exchange_type: &str,
+    durable: bool,
+    auto_delete: bool,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Create exchange").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.create_exchange(&ns, name, exchange_type, durable, auto_delete).await
+}
+
+pub async fn mq_delete_exchange_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    name: &str,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Delete exchange").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.delete_exchange(&ns, name).await
+}
+
+pub async fn mq_list_bindings_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    exchange: Option<String>,
+    queue: Option<String>,
+) -> Result<Vec<MqBindingInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_bindings(&ns, exchange.as_deref(), queue.as_deref()).await
+}
+
+pub async fn mq_bind_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    binding: MqBindingInfo,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Create binding").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.bind_queue(&ns, &binding).await
+}
+
+pub async fn mq_unbind_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    binding: MqBindingInfo,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Delete binding").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.unbind_queue(&ns, &binding).await
+}
+
+// ---- Client connections / channels (RabbitMQ) ----
+
+pub async fn mq_list_client_connections_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+) -> Result<Vec<MqClientConnectionInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_client_connections(&ns).await
+}
+
+pub async fn mq_list_client_channels_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    connection: Option<String>,
+) -> Result<Vec<MqChannelInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_client_channels(&ns, connection).await
+}
+
+pub async fn mq_close_client_connection_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    name: &str,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Close client connection").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.close_client_connection(&ns, name).await
+}
+
+// ---- Users & virtual-host permissions (RabbitMQ) ----
+
+pub async fn mq_list_users_core(state: &AppState, conn_id: &str) -> Result<Vec<MqUserInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_users().await
+}
+
+pub async fn mq_create_user_core(
+    state: &AppState,
+    conn_id: &str,
+    name: &str,
+    password: &str,
+    tags: Vec<String>,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Create user").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.create_user(name, password, tags).await
+}
+
+pub async fn mq_delete_user_core(state: &AppState, conn_id: &str, name: &str) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Delete user").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.delete_user(name).await
+}
+
+pub async fn mq_list_user_permissions_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+) -> Result<Vec<MqVhostPermission>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_user_permissions(&ns).await
+}
+
+pub async fn mq_grant_user_permission_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    user: &str,
+    configure: &str,
+    write: &str,
+    read: &str,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Grant user permission").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.grant_user_permission(&ns, user, configure, write, read).await
+}
+
+pub async fn mq_revoke_user_permission_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    user: &str,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Revoke user permission").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.revoke_user_permission(&ns, user).await
+}
+
+// ---- Policies (RabbitMQ) ----
+
+pub async fn mq_list_policies_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+) -> Result<Vec<MqPolicyInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_policies(&ns).await
+}
+
+pub async fn mq_set_policy_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    policy: MqPolicyInfo,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Set policy").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.set_policy(&ns, &policy).await
+}
+
+pub async fn mq_delete_policy_core(
+    state: &AppState,
+    conn_id: &str,
+    ns: NamespaceRef,
+    name: &str,
+) -> Result<(), String> {
+    ensure_connection_writable(state, conn_id, "Delete policy").await?;
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.delete_policy(&ns, name).await
+}
+
+// ---- Cluster monitoring (RabbitMQ) ----
+
+pub async fn mq_get_overview_core(state: &AppState, conn_id: &str) -> Result<MqOverviewInfo, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.get_overview().await
+}
+
+pub async fn mq_list_nodes_core(state: &AppState, conn_id: &str) -> Result<Vec<MqNodeInfo>, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.list_nodes().await
+}
+
 // ---- Subscriptions ----
 
 pub async fn mq_list_subscriptions_core(
@@ -599,12 +804,13 @@ async fn get_adapter(
     state.mq_registry.get_or_build_config(conn_id, mqc, agent_launch).await
 }
 
-/// Resolve the MQ agent launch spec for agent-backed systems (Kafka, RocketMQ).
+/// Resolve the MQ agent launch spec for agent-backed systems (Kafka, RocketMQ, RabbitMQ).
 /// Returns `None` for native REST systems so the registry skips agent resolution.
 pub fn resolve_mq_agent_launch_spec(mqc: &MqAdminConfig, state: &AppState) -> Option<AgentLaunchSpec> {
     let agent_key = match mqc.system_kind {
         MqSystemKind::Kafka => "kafka",
         MqSystemKind::RocketMq => "rocketmq",
+        MqSystemKind::RabbitMq => "rabbitmq",
         _ => return None,
     };
     let agent_state = state.agent_manager.load_state();
@@ -754,6 +960,19 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    #[tokio::test]
+    async fn mutating_exchange_calls_block_read_only_connections_before_adapter_build() {
+        let (state, dir) = test_state_with(mq_connection(true)).await;
+        let ns = NamespaceRef { tenant: "_rabbitmq".to_string(), namespace: "_rabbitmq".to_string() };
+
+        let err = mq_create_exchange_core(&state, "readonly-mq", ns, "dbx-events", "topic", true, false)
+            .await
+            .expect_err("read-only exchange create should fail");
+
+        assert!(err.contains("Read-only mode"));
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[tokio::test]
     async fn mutating_raw_requests_block_read_only_connections_before_adapter_build() {
         let (state, dir) = test_state_with(mq_connection(true)).await;
@@ -774,6 +993,53 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    #[tokio::test]
+    async fn mutating_user_permission_calls_block_read_only_connections_before_adapter_build() {
+        let (state, dir) = test_state_with(mq_connection(true)).await;
+        let ns = NamespaceRef { tenant: "_rabbitmq".to_string(), namespace: "orders".to_string() };
+
+        let err = mq_create_user_core(&state, "readonly-mq", "dbx-app", "secret", vec![])
+            .await
+            .expect_err("read-only user create should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let err =
+            mq_delete_user_core(&state, "readonly-mq", "dbx-app").await.expect_err("read-only user delete should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let err = mq_grant_user_permission_core(&state, "readonly-mq", ns.clone(), "dbx-app", ".*", ".*", ".*")
+            .await
+            .expect_err("read-only permission grant should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let err = mq_revoke_user_permission_core(&state, "readonly-mq", ns, "dbx-app")
+            .await
+            .expect_err("read-only permission revoke should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[tokio::test]
+    async fn mutating_policy_calls_block_read_only_connections_before_adapter_build() {
+        let (state, dir) = test_state_with(mq_connection(true)).await;
+        let ns = NamespaceRef { tenant: "_rabbitmq".to_string(), namespace: "orders".to_string() };
+        let policy =
+            MqPolicyInfo { name: "dbx-ttl".to_string(), pattern: "^dbx-".to_string(), ..MqPolicyInfo::default() };
+
+        let err = mq_set_policy_core(&state, "readonly-mq", ns.clone(), policy)
+            .await
+            .expect_err("read-only policy set should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let err = mq_delete_policy_core(&state, "readonly-mq", ns, "dbx-ttl")
+            .await
+            .expect_err("read-only policy delete should fail");
+        assert!(err.contains("Read-only mode"));
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[tokio::test]
     async fn peek_messages_rejects_counts_above_service_limit_before_adapter_call() {
         let (state, dir) = test_state_with(mq_connection(false)).await;

crates/dbx-core/src/mq/types.rs

@@ -14,6 +14,8 @@ pub enum MqSystemKind {
     Kafka,
     #[serde(rename = "rocketmq")]
     RocketMq,
+    #[serde(rename = "rabbitmq")]
+    RabbitMq,
 }
 
 impl MqSystemKind {
@@ -22,6 +24,7 @@ impl MqSystemKind {
             MqSystemKind::Pulsar => "pulsar",
             MqSystemKind::Kafka => "kafka",
             MqSystemKind::RocketMq => "rocketmq",
+            MqSystemKind::RabbitMq => "rabbitmq",
         }
     }
 }
@@ -61,6 +64,21 @@ pub struct MqCapabilities {
     /// RocketMQ: message trace lookup (requires broker trace topic).
     #[serde(default)]
     pub supports_message_trace: bool,
+    /// RabbitMQ: exchange & binding management.
+    #[serde(default)]
+    pub supports_exchanges: bool,
+    /// RabbitMQ: client connection & channel management (list/close).
+    #[serde(default)]
+    pub supports_client_connections: bool,
+    /// RabbitMQ: user & virtual-host permission management.
+    #[serde(default)]
+    pub supports_user_permissions: bool,
+    /// RabbitMQ: policy management (list/set/delete policies per vhost).
+    #[serde(default)]
+    pub supports_policies: bool,
+    /// RabbitMQ: cluster overview & node monitoring via the management API.
+    #[serde(default)]
+    pub supports_cluster_monitoring: bool,
 }
 
 /// Result of a connectivity test, including the detected server version and how
@@ -309,6 +327,10 @@ pub struct TopicInfo {
     /// RocketMQ message type from broker topic config (NORMAL, DELAY, FIFO, etc.).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub message_type: Option<String>,
+    /// Namespace (RabbitMQ: virtual host) this item belongs to; set on
+    /// cross-namespace listings such as the RabbitMQ "all vhosts" mode.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
 }
 
 #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
@@ -638,6 +660,218 @@ pub struct MqRawResponse {
     pub text: Option<String>,
 }
 
+// ---------------------------------------------------------------------------
+// Exchange / Binding (RabbitMQ)
+// ---------------------------------------------------------------------------
+
+/// A RabbitMQ exchange. Namespaces map to virtual hosts, so the exchange's
+/// vhost is normally carried by the `NamespaceRef` passed alongside it; in
+/// "all vhosts" listings the per-item vhost is reported via `namespace`.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqExchangeInfo {
+    pub name: String,
+    /// `direct` | `fanout` | `topic` | `headers`.
+    #[serde(rename = "type")]
+    pub exchange_type: String,
+    #[serde(default)]
+    pub durable: bool,
+    #[serde(default)]
+    pub auto_delete: bool,
+    /// Internal exchange (`amq.*`); cannot be deleted and is hidden by default in the UI.
+    #[serde(default)]
+    pub internal: bool,
+    /// Virtual host this exchange belongs to, set on all-vhosts listings.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
+}
+
+/// A RabbitMQ binding between an exchange (source) and a queue or another
+/// exchange (destination).
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqBindingInfo {
+    /// Source exchange name.
+    pub source: String,
+    /// Destination queue or exchange name.
+    pub destination: String,
+    /// `queue` | `exchange`.
+    pub destination_type: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub routing_key: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub arguments: Option<HashMap<String, serde_json::Value>>,
+    /// Virtual host this binding belongs to, set on all-vhosts listings.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
+}
+
+// ---------------------------------------------------------------------------
+// Client connections / channels (RabbitMQ)
+// ---------------------------------------------------------------------------
+
+/// A RabbitMQ client connection as reported by the management API. Virtual
+/// host scoping is normally carried by the `NamespaceRef` passed alongside
+/// it; in "all vhosts" listings the per-item vhost is reported via
+/// `namespace`.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqClientConnectionInfo {
+    /// Server-side connection name (`host:port -> host:port`).
+    pub name: String,
+    /// Authenticated username.
+    #[serde(default)]
+    pub user: String,
+    #[serde(default)]
+    pub peer_host: String,
+    #[serde(default)]
+    pub peer_port: i32,
+    /// `running` | `blocked` | `blocking` | ...
+    #[serde(default)]
+    pub state: String,
+    /// Number of channels open on this connection.
+    #[serde(default)]
+    pub channels: u32,
+    /// Receive rate (bytes/s), when the management API reports it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub recv_rate: Option<f64>,
+    /// Send rate (bytes/s), when the management API reports it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub send_rate: Option<f64>,
+    /// Connection establishment time (epoch milliseconds), when reported.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub connected_at: Option<i64>,
+    /// Virtual host this connection is attached to, set on all-vhosts listings.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
+}
+
+/// A RabbitMQ channel as reported by the management API.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqChannelInfo {
+    /// Channel name (`<connection name> (<channel number>)`).
+    pub name: String,
+    /// Name of the connection this channel belongs to, when reported.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub connection_name: Option<String>,
+    #[serde(default)]
+    pub state: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub prefetch: Option<u32>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub messages_unacked: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub consumer_count: Option<u32>,
+    /// Virtual host this channel belongs to, set on all-vhosts listings.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
+}
+
+// ---------------------------------------------------------------------------
+// Users & virtual-host permissions (RabbitMQ)
+// ---------------------------------------------------------------------------
+
+/// A RabbitMQ user account as reported by the management API.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqUserInfo {
+    pub name: String,
+    #[serde(default)]
+    pub tags: Vec<String>,
+}
+
+/// A RabbitMQ user × virtual host permission triple: the `configure` / `write`
+/// / `read` regexes scoped to one virtual host.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqVhostPermission {
+    pub user: String,
+    pub vhost: String,
+    pub configure: String,
+    pub write: String,
+    pub read: String,
+}
+
+// ---------------------------------------------------------------------------
+// Policies & cluster monitoring (RabbitMQ)
+// ---------------------------------------------------------------------------
+
+/// A RabbitMQ policy as reported by the management API. Unlike exchanges and
+/// bindings, policies always carry their virtual host explicitly (`vhost`),
+/// including on single-vhost listings.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqPolicyInfo {
+    pub name: String,
+    #[serde(default)]
+    pub vhost: String,
+    /// Regex matching the queues/exchanges this policy applies to.
+    #[serde(default)]
+    pub pattern: String,
+    /// `queues` | `exchanges` | `all`.
+    #[serde(default, rename = "applyTo")]
+    pub apply_to: String,
+    #[serde(default)]
+    pub priority: i32,
+    /// Policy key/value pairs (`max-length`, `message-ttl`, ...).
+    #[serde(default)]
+    pub definition: HashMap<String, serde_json::Value>,
+}
+
+/// Broker-wide queue totals and message rates from the RabbitMQ management
+/// API `overview` endpoint. All fields are optional because the agent omits
+/// values the broker does not report.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqOverviewInfo {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub messages_ready: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub messages_unacked: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub publish_rate: Option<f64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub deliver_rate: Option<f64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ack_rate: Option<f64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_queues: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_exchanges: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_connections: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_channels: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_consumers: Option<i64>,
+}
+
+/// One RabbitMQ cluster node as reported by the management API.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqNodeInfo {
+    pub name: String,
+    #[serde(default)]
+    pub running: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mem_used: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mem_limit: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub disk_free: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub fd_used: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub fd_total: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sockets_used: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sockets_total: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub uptime_ms: Option<i64>,
+}
+
 // ---------------------------------------------------------------------------
 // Send message (produce)
 // ---------------------------------------------------------------------------
@@ -666,6 +900,16 @@ pub struct SendMessageRequest {
     /// is used.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub partition: Option<i32>,
+    /// RabbitMQ: target exchange. When omitted, the agent publishes to the
+    /// default exchange with the queue name as routing key.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub exchange: Option<String>,
+    /// RabbitMQ: routing key used together with `exchange`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub routing_key: Option<String>,
+    /// RabbitMQ: namespace hint that maps to the target virtual host.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub namespace: Option<String>,
 }
 
 /// Result of a successful message production.
@@ -697,4 +941,254 @@ mod tests {
             .expect("message-id reset position");
         assert!(matches!(pos, ResetPosition::MessageId { ledger_id: 5, entry_id: 9 }));
     }
+
+    #[test]
+    fn exchange_info_round_trips_agent_camel_case() {
+        let exchange: super::MqExchangeInfo = serde_json::from_str(
+            r#"{"name":"dbx-events","type":"topic","durable":true,"autoDelete":false,"internal":false}"#,
+        )
+        .expect("exchange info");
+        assert_eq!(exchange.name, "dbx-events");
+        assert_eq!(exchange.exchange_type, "topic");
+        assert!(exchange.durable);
+        assert!(!exchange.auto_delete);
+        assert!(!exchange.internal);
+
+        let json = serde_json::to_value(&exchange).expect("serialize exchange info");
+        assert_eq!(json.get("type").and_then(|v| v.as_str()), Some("topic"));
+        assert_eq!(json.get("autoDelete").and_then(|v| v.as_bool()), Some(false));
+    }
+
+    #[test]
+    fn binding_info_skips_absent_routing_key_and_arguments() {
+        let binding: super::MqBindingInfo = serde_json::from_str(
+            r#"{"source":"dbx-events","destination":"dbx-queue","destinationType":"queue","routingKey":"orders.*"}"#,
+        )
+        .expect("binding info");
+        assert_eq!(binding.source, "dbx-events");
+        assert_eq!(binding.destination_type, "queue");
+        assert_eq!(binding.routing_key.as_deref(), Some("orders.*"));
+        assert!(binding.arguments.is_none());
+
+        let json = serde_json::to_value(&binding).expect("serialize binding info");
+        assert!(json.get("arguments").is_none());
+
+        let bare: super::MqBindingInfo =
+            serde_json::from_str(r#"{"source":"dbx-a","destination":"dbx-b","destinationType":"exchange"}"#)
+                .expect("binding without routing key");
+        let json = serde_json::to_value(&bare).expect("serialize bare binding");
+        assert!(json.get("routingKey").is_none());
+        assert!(json.get("arguments").is_none());
+    }
+
+    #[test]
+    fn send_message_request_defaults_new_rabbitmq_fields() {
+        let req: super::SendMessageRequest =
+            serde_json::from_str(r#"{"topic":"dbx-queue","payloadBase64":"aGVsbG8="}"#).expect("send request");
+        assert!(req.exchange.is_none());
+        assert!(req.routing_key.is_none());
+        assert!(req.namespace.is_none());
+    }
+
+    #[test]
+    fn client_connection_info_round_trips_agent_camel_case() {
+        let conn: super::MqClientConnectionInfo = serde_json::from_str(
+            r#"{"name":"192.168.1.10:52344 -> 192.168.1.126:5672","user":"jjsd","peerHost":"192.168.1.10","peerPort":52344,"state":"running","channels":3,"recvRate":12.5,"sendRate":34.0,"connectedAt":1710000000000}"#,
+        )
+        .expect("client connection info");
+        assert_eq!(conn.name, "192.168.1.10:52344 -> 192.168.1.126:5672");
+        assert_eq!(conn.user, "jjsd");
+        assert_eq!(conn.peer_host, "192.168.1.10");
+        assert_eq!(conn.peer_port, 52344);
+        assert_eq!(conn.state, "running");
+        assert_eq!(conn.channels, 3);
+        assert_eq!(conn.recv_rate, Some(12.5));
+        assert_eq!(conn.send_rate, Some(34.0));
+        assert_eq!(conn.connected_at, Some(1710000000000));
+
+        let json = serde_json::to_value(&conn).expect("serialize client connection info");
+        assert_eq!(json.get("peerHost").and_then(|v| v.as_str()), Some("192.168.1.10"));
+        assert_eq!(json.get("peerPort").and_then(|v| v.as_i64()), Some(52344));
+        assert_eq!(json.get("connectedAt").and_then(|v| v.as_i64()), Some(1710000000000));
+    }
+
+    #[test]
+    fn client_connection_info_skips_absent_optional_fields() {
+        let conn: super::MqClientConnectionInfo = serde_json::from_str(
+            r#"{"name":"a:1 -> b:5672","user":"guest","peerHost":"a","peerPort":1,"state":"running","channels":0}"#,
+        )
+        .expect("client connection info without rates");
+        assert!(conn.recv_rate.is_none());
+        assert!(conn.send_rate.is_none());
+        assert!(conn.connected_at.is_none());
+
+        let json = serde_json::to_value(&conn).expect("serialize client connection info");
+        assert!(json.get("recvRate").is_none());
+        assert!(json.get("sendRate").is_none());
+        assert!(json.get("connectedAt").is_none());
+    }
+
+    #[test]
+    fn user_info_round_trips_agent_camel_case() {
+        let user: super::MqUserInfo =
+            serde_json::from_str(r#"{"name":"dbx-app","tags":["management","policymaker"]}"#).expect("user info");
+        assert_eq!(user.name, "dbx-app");
+        assert_eq!(user.tags, vec!["management".to_string(), "policymaker".to_string()]);
+
+        // Users without tags default to an empty list.
+        let bare: super::MqUserInfo = serde_json::from_str(r#"{"name":"dbx-svc"}"#).expect("user without tags");
+        assert!(bare.tags.is_empty());
+
+        let json = serde_json::to_value(&user).expect("serialize user info");
+        assert_eq!(json.get("name").and_then(|v| v.as_str()), Some("dbx-app"));
+        assert_eq!(json.get("tags").and_then(|v| v.as_array()).map(|a| a.len()), Some(2));
+    }
+
+    #[test]
+    fn vhost_permission_round_trips_agent_fields() {
+        let perm: super::MqVhostPermission = serde_json::from_str(
+            r#"{"user":"dbx-app","vhost":"orders","configure":".*","write":"dbx-.*","read":".*"}"#,
+        )
+        .expect("vhost permission");
+        assert_eq!(perm.user, "dbx-app");
+        assert_eq!(perm.vhost, "orders");
+        assert_eq!(perm.configure, ".*");
+        assert_eq!(perm.write, "dbx-.*");
+        assert_eq!(perm.read, ".*");
+
+        let json = serde_json::to_value(&perm).expect("serialize vhost permission");
+        assert_eq!(json.get("vhost").and_then(|v| v.as_str()), Some("orders"));
+        assert_eq!(json.get("write").and_then(|v| v.as_str()), Some("dbx-.*"));
+    }
+
+    #[test]
+    fn capabilities_default_user_permissions_off() {
+        // The older capability fields are not `serde(default)`, so a payload
+        // from an adapter predating the new flag is simulated by serializing a
+        // full struct and stripping the new key: it must deserialize with the
+        // flag off.
+        let caps = super::MqCapabilities { supports_tenants: true, ..Default::default() };
+        let mut json = serde_json::to_value(caps).expect("serialize capabilities");
+        assert_eq!(json.get("supportsUserPermissions").and_then(|v| v.as_bool()), Some(false));
+        json.as_object_mut().expect("capabilities object").remove("supportsUserPermissions");
+
+        let caps: super::MqCapabilities = serde_json::from_value(json).expect("deserialize without the new field");
+        assert!(!caps.supports_user_permissions);
+        assert!(caps.supports_tenants);
+    }
+
+    #[test]
+    fn channel_info_round_trips_agent_camel_case() {
+        let channel: super::MqChannelInfo = serde_json::from_str(
+            r#"{"name":"a:1 -> b:5672 (1)","connectionName":"a:1 -> b:5672","state":"running","prefetch":20,"messagesUnacked":7,"consumerCount":2}"#,
+        )
+        .expect("channel info");
+        assert_eq!(channel.name, "a:1 -> b:5672 (1)");
+        assert_eq!(channel.connection_name.as_deref(), Some("a:1 -> b:5672"));
+        assert_eq!(channel.state, "running");
+        assert_eq!(channel.prefetch, Some(20));
+        assert_eq!(channel.messages_unacked, Some(7));
+        assert_eq!(channel.consumer_count, Some(2));
+
+        let json = serde_json::to_value(&channel).expect("serialize channel info");
+        assert_eq!(json.get("connectionName").and_then(|v| v.as_str()), Some("a:1 -> b:5672"));
+        assert_eq!(json.get("messagesUnacked").and_then(|v| v.as_u64()), Some(7));
+        assert_eq!(json.get("consumerCount").and_then(|v| v.as_u64()), Some(2));
+
+        let bare: super::MqChannelInfo =
+            serde_json::from_str(r#"{"name":"a:1 -> b:5672 (2)","state":"running"}"#).expect("bare channel info");
+        let json = serde_json::to_value(&bare).expect("serialize bare channel info");
+        assert!(json.get("connectionName").is_none());
+        assert!(json.get("prefetch").is_none());
+        assert!(json.get("messagesUnacked").is_none());
+        assert!(json.get("consumerCount").is_none());
+    }
+
+    #[test]
+    fn policy_info_round_trips_agent_fields() {
+        let policy: super::MqPolicyInfo = serde_json::from_str(
+            r#"{"name":"dbx-ttl","vhost":"orders","pattern":"^dbx-","applyTo":"queues","priority":5,"definition":{"message-ttl":60000,"max-length":1000}}"#,
+        )
+        .expect("policy info");
+        assert_eq!(policy.name, "dbx-ttl");
+        assert_eq!(policy.vhost, "orders");
+        assert_eq!(policy.pattern, "^dbx-");
+        assert_eq!(policy.apply_to, "queues");
+        assert_eq!(policy.priority, 5);
+        assert_eq!(policy.definition.get("message-ttl").and_then(|v| v.as_i64()), Some(60000));
+
+        let json = serde_json::to_value(&policy).expect("serialize policy info");
+        assert_eq!(json.get("applyTo").and_then(|v| v.as_str()), Some("queues"));
+        assert_eq!(json.get("vhost").and_then(|v| v.as_str()), Some("orders"));
+    }
+
+    #[test]
+    fn policy_info_defaults_absent_apply_to_and_priority() {
+        let policy: super::MqPolicyInfo =
+            serde_json::from_str(r#"{"name":"dbx-ha","vhost":"/","pattern":".*","definition":{"ha-mode":"all"}}"#)
+                .expect("policy without applyTo/priority");
+        assert!(policy.apply_to.is_empty());
+        assert_eq!(policy.priority, 0);
+        assert_eq!(policy.definition.get("ha-mode").and_then(|v| v.as_str()), Some("all"));
+    }
+
+    #[test]
+    fn overview_info_skips_absent_optional_fields() {
+        let overview: super::MqOverviewInfo = serde_json::from_str(
+            r#"{"messagesReady":12,"messagesUnacked":3,"publishRate":1.5,"deliverRate":2.0,"ackRate":2.0,"totalQueues":4,"totalExchanges":7,"totalConnections":2,"totalChannels":5,"totalConsumers":6}"#,
+        )
+        .expect("overview info");
+        assert_eq!(overview.messages_ready, Some(12));
+        assert_eq!(overview.publish_rate, Some(1.5));
+        assert_eq!(overview.total_consumers, Some(6));
+
+        let json = serde_json::to_value(&overview).expect("serialize overview info");
+        assert_eq!(json.get("messagesReady").and_then(|v| v.as_i64()), Some(12));
+        assert_eq!(json.get("ackRate").and_then(|v| v.as_f64()), Some(2.0));
+
+        let bare: super::MqOverviewInfo = serde_json::from_str(r#"{}"#).expect("empty overview");
+        assert!(bare.messages_ready.is_none());
+        let json = serde_json::to_value(&bare).expect("serialize empty overview");
+        assert!(json.get("messagesReady").is_none());
+        assert!(json.get("totalQueues").is_none());
+    }
+
+    #[test]
+    fn node_info_round_trips_agent_camel_case() {
+        let node: super::MqNodeInfo = serde_json::from_str(
+            r#"{"name":"rabbit@node1","running":true,"memUsed":1024,"memLimit":2048,"diskFree":4096,"fdUsed":10,"fdTotal":100,"socketsUsed":3,"socketsTotal":50,"uptimeMs":1710000000000}"#,
+        )
+        .expect("node info");
+        assert_eq!(node.name, "rabbit@node1");
+        assert!(node.running);
+        assert_eq!(node.mem_used, Some(1024));
+        assert_eq!(node.fd_total, Some(100));
+        assert_eq!(node.uptime_ms, Some(1710000000000));
+
+        let json = serde_json::to_value(&node).expect("serialize node info");
+        assert_eq!(json.get("memUsed").and_then(|v| v.as_i64()), Some(1024));
+        assert_eq!(json.get("socketsTotal").and_then(|v| v.as_i64()), Some(50));
+
+        let bare: super::MqNodeInfo =
+            serde_json::from_str(r#"{"name":"rabbit@node2","running":false}"#).expect("bare node info");
+        assert!(!bare.running);
+        let json = serde_json::to_value(&bare).expect("serialize bare node info");
+        assert!(json.get("memUsed").is_none());
+        assert!(json.get("uptimeMs").is_none());
+    }
+
+    #[test]
+    fn capabilities_default_policies_and_cluster_monitoring_off() {
+        // Adapters predating the new flags omit them; deserialization must
+        // default both to off.
+        let caps = super::MqCapabilities { supports_tenants: true, ..Default::default() };
+        let mut json = serde_json::to_value(caps).expect("serialize capabilities");
+        json.as_object_mut().expect("capabilities object").remove("supportsPolicies");
+        json.as_object_mut().expect("capabilities object").remove("supportsClusterMonitoring");
+
+        let caps: super::MqCapabilities = serde_json::from_value(json).expect("deserialize without the new fields");
+        assert!(!caps.supports_policies);
+        assert!(!caps.supports_cluster_monitoring);
+        assert!(caps.supports_tenants);
+    }
 }