feat(rabbitmq): replace Java agent with Go implementation

.github/scripts/bump-agent-versions.mjs

@@ -46,6 +46,7 @@ const nativeDriverDirectories = {
   duckdb: "duckdb",
   oracle: "oracle-go",
   kingbase: "kingbase-go",
+  rabbitmq: "rabbitmq",
 };
 
 function resolveAgentModule(moduleName, { legacyStandaloneModules, moduleExists, readModuleFile }) {

.github/scripts/bump-agent-versions.test.mjs

@@ -29,3 +29,14 @@ test("bumps DuckDB after its initial release", () => {
 
   assert.equal(result.versions.duckdb, "0.1.1");
 });
+
+test("bumps the native RabbitMQ agent from its Go directory", () => {
+  const result = evaluateAgentVersionBump({
+    versions: { rabbitmq: "0.1.0" },
+    changedFiles: ["agents/drivers/rabbitmq/main.go"],
+    moduleExists: (path) => path === "agents/drivers/rabbitmq",
+    readModuleFile: () => "",
+  });
+
+  assert.equal(result.versions.rabbitmq, "0.1.1");
+});

.github/workflows/agents-release.yml

@@ -217,6 +217,45 @@ jobs:
           name: xugu-native
           path: "release-native/dbx-agent-xugu-*"
 
+  build-rabbitmq-native:
+    needs: [bump-versions]
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "1.22.x"
+      - name: Test RabbitMQ native agent
+        working-directory: agents/drivers/rabbitmq
+        run: go test ./...
+      - name: Cross-compile RabbitMQ native agent
+        shell: bash
+        run: |
+          mkdir -p release-native
+          cd agents/drivers/rabbitmq
+          declare -A TARGETS=(
+            ["macos-aarch64"]="darwin/arm64"
+            ["macos-x64"]="darwin/amd64"
+            ["linux-aarch64"]="linux/arm64"
+            ["linux-x64"]="linux/amd64"
+            ["windows-aarch64"]="windows/arm64"
+            ["windows-x64"]="windows/amd64"
+          )
+          for platform in "${!TARGETS[@]}"; do
+            IFS=/ read -r goos goarch <<< "${TARGETS[$platform]}"
+            output="../../../release-native/dbx-agent-rabbitmq-${platform}"
+            if [[ "$goos" == "windows" ]]; then
+              output="${output}.exe"
+            fi
+            echo "Building $platform ($goos/$goarch)"
+            CGO_ENABLED=0 GOOS="$goos" GOARCH="$goarch" go build -trimpath -ldflags="-s -w" -o "$output" .
+          done
+          ls -lh ../../../release-native
+      - uses: actions/upload-artifact@v4
+        with:
+          name: rabbitmq-native
+          path: "release-native/dbx-agent-rabbitmq-*"
+
   build-kingbase-native:
     needs: [bump-versions]
     runs-on: ubuntu-latest
@@ -436,7 +475,7 @@ jobs:
           path: "dbx-jre-*.tar.zst"
 
   release:
-    needs: [bump-versions, commit-versions, build-agents, build-oracle-native, build-xugu-native, build-kingbase-native, build-duckdb-native, build-jre]
+    needs: [bump-versions, commit-versions, build-agents, build-oracle-native, build-xugu-native, build-rabbitmq-native, build-kingbase-native, build-duckdb-native, build-jre]
     runs-on: ubuntu-latest
     steps:
       - name: Create DBX bot release token
@@ -466,6 +505,7 @@ jobs:
           find artifacts/agent-jars -name '*.jar' -exec cp {} release/ \;
           find artifacts/oracle-native -type f -name 'dbx-agent-oracle-*' -exec cp {} release/ \;
           find artifacts/xugu-native -type f -name 'dbx-agent-xugu-*' -exec cp {} release/ \;
+          find artifacts/rabbitmq-native -type f -name 'dbx-agent-rabbitmq-*' -exec cp {} release/ \;
           find artifacts/kingbase-native -type f -name 'dbx-agent-kingbase-*' -exec cp {} release/ \;
           find artifacts/duckdb-native-* -type f -name 'dbx-agent-duckdb-*' -exec cp {} release/ \;
           find artifacts -name 'dbx-jre-*.tar.zst' -exec cp {} release/ \;
@@ -540,6 +580,7 @@ jobs:
               kingbase) echo "人大金仓 KingbaseES" ;;
               duckdb) echo "DuckDB" ;;
               xugu) echo "虚谷 XuguDB" ;;
+              rabbitmq) echo "RabbitMQ" ;;
               *) echo "$name" ;;
             esac
           }
@@ -603,7 +644,7 @@ jobs:
             [ -n "$DRIVERS" ] && DRIVERS="${DRIVERS},"$'\n'
             DRIVERS="${DRIVERS}$(generate_jar_entry "$name" "$label" "$f" "$jre_key" "$version" "$external_driver" "$native_json")"
           done
-          for name in oracle xugu kingbase duckdb; do
+          for name in oracle xugu kingbase duckdb rabbitmq; do
             version=$(get_module_version "$name")
             [ -f "release/dbx-agent-${name}-${version}.jar" ] && continue
             native_json=$(generate_native_platforms "$name" "$version")
@@ -673,6 +714,7 @@ jobs:
               duckdb) echo "DuckDB" ;;
               oracle) echo "Oracle" ;;
               xugu) echo "虚谷 XuguDB" ;;
+              rabbitmq) echo "RabbitMQ" ;;
               *) echo "$name" ;;
             esac
           }

.github/workflows/ci.yml

@@ -513,6 +513,10 @@ jobs:
         run: GONOSUMDB=gitee.com/XuguDB/go-xugu-driver go test ./...
         working-directory: agents/drivers/xugu
 
+      - name: RabbitMQ native agent tests
+        run: go test ./...
+        working-directory: agents/drivers/rabbitmq
+
       - name: Oracle native agent build
         run: CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /tmp/dbx-agent-oracle-linux-x64 .
         working-directory: agents/drivers/oracle-go
@@ -521,6 +525,46 @@ jobs:
         run: GONOSUMDB=gitee.com/XuguDB/go-xugu-driver CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /tmp/dbx-agent-xugu-linux-x64 .
         working-directory: agents/drivers/xugu
 
+      - name: RabbitMQ native agent build
+        run: CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /tmp/dbx-agent-rabbitmq-linux-x64 .
+        working-directory: agents/drivers/rabbitmq
+
+      - name: RabbitMQ native agent integration tests
+        shell: bash
+        working-directory: agents/drivers/rabbitmq
+        run: |
+          set -euo pipefail
+          for version in 3.13 4.3; do
+            name="dbx-rabbitmq-${version//./-}"
+            docker run -d --name "$name" \
+              -e RABBITMQ_DEFAULT_USER=dbx \
+              -e RABBITMQ_DEFAULT_PASS=dbx-password \
+              -p 5672:5672 -p 15672:15672 \
+              "rabbitmq:${version}-management"
+            cleanup() {
+              docker rm -f "$name" >/dev/null 2>&1 || true
+            }
+            trap cleanup EXIT
+            ready=false
+            for _ in $(seq 1 60); do
+              if docker exec "$name" rabbitmq-diagnostics -q ping >/dev/null 2>&1; then
+                ready=true
+                break
+              fi
+              sleep 2
+            done
+            if [ "$ready" != "true" ]; then
+              docker logs "$name"
+              exit 1
+            fi
+            RABBITMQ_INTEGRATION=1 \
+              RABBITMQ_USERNAME=dbx \
+              RABBITMQ_PASSWORD=dbx-password \
+              go test -run '^TestRabbitMQIntegration$' -count=1 ./...
+            cleanup
+            trap - EXIT
+          done
+
       - name: Java agent tests and packages
         run: ./gradlew test shadowJar --continue
 

agents/README.md

@@ -44,12 +44,12 @@ Each agent runs as a standalone process and communicates with DBX via stdin/stdo
 | iotdb | Apache IoTDB | IoTDB JDBC |
 | etcd | etcd | jetcd |
 | zookeeper | Apache ZooKeeper | Apache Curator |
-| rabbitmq | RabbitMQ | RabbitMQ AMQP Java client |
+| rabbitmq | RabbitMQ | amqp091-go native agent |
 
 
 ## Multi-JRE Support
 
-Most Java agents target JRE 21. Native agents, such as `duckdb`, `oracle`, `kingbase`, and `xugu`, do not require a JRE. DBX downloads and manages the JRE 21 installation automatically for Java agents.
+Most Java agents target JRE 21. Native agents, such as `duckdb`, `oracle`, `kingbase`, `xugu`, and `rabbitmq`, do not require a JRE. DBX downloads and manages the JRE 21 installation automatically for Java agents.
 
 ## JDBC Connection Pooling
 
@@ -75,7 +75,7 @@ Set `DBX_AGENT_JDBC_POOL_ENABLED=false` for a runtime-level compatibility fallba
 
 For new agents, prefer a **native (Go or Rust) driver** over a Java/JDBC agent whenever a mature, license-compatible native driver is available. Native agents ship as a single self-contained executable with no JRE, which significantly reduces memory footprint and startup time — the JVM baseline that every Java agent pays even when idle is avoided entirely.
 
-- **Native (C++/Go/Rust)** — preferred when a usable native driver exists. See `drivers/duckdb`, `drivers/oracle-go` (go-ora), `drivers/kingbase-go` (gokb), and `drivers/xugu` as reference implementations. No JRE download or management is needed.
+- **Native (C++/Go/Rust)** — preferred when a usable native driver exists. See `drivers/duckdb`, `drivers/oracle-go` (go-ora), `drivers/kingbase-go` (gokb), `drivers/xugu`, and `drivers/rabbitmq` (amqp091-go) as reference implementations. No JRE download or management is needed.
 - **Java/JDBC** — the default fallback when only a JDBC driver exists for the database, or when the native driver is immature or unmaintained. Most agents still fall in this category.
 
 Native agents implement the same JSON-RPC contract and `versions.json` registration as Java agents; they ship an `agent` executable instead of `agent.jar`. If both native and Java source implementations exist for the same database, publish only the native artifact unless the Java variant has a separately registered compatibility profile, such as `oracle-legacy` / `oracle-10g`.
@@ -89,9 +89,10 @@ Requires JDK 21 (Gradle toolchain auto-downloads if needed).
 (cd drivers/oracle-go && go build -o agent .)
 (cd drivers/kingbase-go && go build -o agent .)
 (cd drivers/xugu && go build -o agent .)
+(cd drivers/rabbitmq && go build -o agent .)
 ```
 
-Output JARs are in `drivers/{module}/build/libs/`. Native agents build from `drivers/oracle-go`, `drivers/kingbase-go`, and `drivers/xugu`.
+Output JARs are in `drivers/{module}/build/libs/`. Native agents build from `drivers/oracle-go`, `drivers/kingbase-go`, `drivers/xugu`, and `drivers/rabbitmq`.
 
 ### Local DBX Runtime Test
 
@@ -105,7 +106,7 @@ cp agents/drivers/<db_type>/build/libs/*-all.jar ~/.dbx/agents/drivers/<db_type>
 
 Restart DBX or disconnect and reconnect the database so the new agent process loads the replacement JAR.
 
-Native agents such as `oracle`, `kingbase`, and `xugu` use the `agent` executable in the driver directory instead of `agent.jar`.
+Native agents such as `oracle`, `kingbase`, `xugu`, and `rabbitmq` use the `agent` executable in the driver directory instead of `agent.jar`.
 
 ## Versioning
 

agents/README.zh-CN.md

@@ -44,12 +44,12 @@ DBX 的 Agent 驱动 —— 通过 JDBC 和原生数据库驱动支持各种数
 | iotdb | Apache IoTDB | IoTDB JDBC |
 | etcd | etcd | jetcd |
 | zookeeper | Apache ZooKeeper | Apache Curator |
-| rabbitmq | RabbitMQ | RabbitMQ AMQP Java client |
+| rabbitmq | RabbitMQ | amqp091-go 原生 agent |
 
 
 ## 多 JRE 支持
 
-多数 Java agent 以 JRE 21 为目标。原生 agent（如 `oracle`、`kingbase` 和 `xugu`）不需要 JRE。对 Java agent，DBX 会自动下载并管理 JRE 21 安装。
+多数 Java agent 以 JRE 21 为目标。原生 agent（如 `oracle`、`kingbase`、`xugu` 和 `rabbitmq`）不需要 JRE。对 Java agent，DBX 会自动下载并管理 JRE 21 安装。
 
 ## JDBC 连接池
 
@@ -75,7 +75,7 @@ HikariCP 会直接打进启用连接池的 Agent JAR。已经使用 DBX 托管 J
 
 对于新 agent，只要存在成熟、许可证兼容的原生驱动，优先选择**原生（Go 或 Rust）驱动**而非 Java/JDBC agent。原生 agent 以单一自包含可执行文件发布，无需 JRE，可显著降低内存占用和启动时间 —— 完全避开 Java agent 即便空闲也要付出的 JVM 基线开销。
 
-- **原生（Go/Rust）** —— 存在可用原生驱动时首选。参考 `drivers/oracle-go`（go-ora）、`drivers/kingbase-go`（gokb）和 `drivers/xugu`。无需 JRE 下载与管理。
+- **原生（Go/Rust）** —— 存在可用原生驱动时首选。参考 `drivers/oracle-go`（go-ora）、`drivers/kingbase-go`（gokb）、`drivers/xugu` 和 `drivers/rabbitmq`（amqp091-go）。无需 JRE 下载与管理。
 - **Java/JDBC** —— 当某数据库只有 JDBC 驱动，或原生驱动不成熟、缺乏维护时的默认兜底方案。多数 agent 仍属此类。
 
 原生 agent 实现与 Java agent 相同的 JSON-RPC 契约和 `versions.json` 登记；它发布的是 `agent` 可执行文件而非 `agent.jar`。若同一数据库同时保留原生和 Java 源码实现，默认只发布原生产物；只有 Java 变体以独立兼容配置登记时才同时发布，例如 `oracle-legacy` / `oracle-10g`。
@@ -89,9 +89,10 @@ HikariCP 会直接打进启用连接池的 Agent JAR。已经使用 DBX 托管 J
 (cd drivers/oracle-go && go build -o agent .)
 (cd drivers/kingbase-go && go build -o agent .)
 (cd drivers/xugu && go build -o agent .)
+(cd drivers/rabbitmq && go build -o agent .)
 ```
 
-产物 JAR 在 `drivers/{module}/build/libs/`。原生 agent 从 `drivers/oracle-go`、`drivers/kingbase-go` 和 `drivers/xugu` 构建。
+产物 JAR 在 `drivers/{module}/build/libs/`。原生 agent 从 `drivers/oracle-go`、`drivers/kingbase-go`、`drivers/xugu` 和 `drivers/rabbitmq` 构建。
 
 ### 本地 DBX 运行时测试
 
@@ -105,7 +106,7 @@ cp agents/drivers/<db_type>/build/libs/*-all.jar ~/.dbx/agents/drivers/<db_type>
 
 重启 DBX 或断开重连数据库，使新 agent 进程加载替换后的 JAR。
 
-`oracle`、`kingbase` 和 `xugu` 等原生 agent 使用驱动目录下的 `agent` 可执行文件而非 `agent.jar`。
+`oracle`、`kingbase`、`xugu` 和 `rabbitmq` 等原生 agent 使用驱动目录下的 `agent` 可执行文件而非 `agent.jar`。
 
 ## 版本管理
 

agents/build.gradle

@@ -3,7 +3,7 @@ plugins {
 }
 
 def infrastructureProjects = ['common', 'test-support'] as Set
-def legacyStandaloneProjects = ['mongodb', 'kafka', 'rocketmq', 'rabbitmq'] as Set
+def legacyStandaloneProjects = ['mongodb', 'kafka', 'rocketmq'] as Set
 def pooledJdbcProjects = [
     'access', 'bigquery', 'cassandra', 'dameng', 'databend', 'databricks', 'db2', 'exasol',
     'firebird', 'gbase8a', 'gbase8s', 'goldendb', 'h2', 'h2-legacy', 'highgo', 'hive',

agents/drivers/rabbitmq/bench/agent_compare.go

@@ -0,0 +1,523 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"runtime"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type agentSpec struct {
+	Name         string
+	Command      []string
+	ArtifactPath string
+}
+
+type agentProcess struct {
+	command *exec.Cmd
+	stdin   io.WriteCloser
+	reader  *bufio.Scanner
+	nextID  int64
+}
+
+type agentResponse struct {
+	ID     int64           `json:"id"`
+	Result json.RawMessage `json:"result"`
+	Error  *struct {
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+type benchmarkResult struct {
+	Agent         string  `json:"agent"`
+	Workload      string  `json:"workload"`
+	Round         int     `json:"round"`
+	Operations    int     `json:"operations"`
+	Errors        int     `json:"errors"`
+	DurationMS    float64 `json:"duration_ms"`
+	QPS           float64 `json:"qps"`
+	MeanMS        float64 `json:"mean_ms"`
+	P50MS         float64 `json:"p50_ms"`
+	P95MS         float64 `json:"p95_ms"`
+	P99MS         float64 `json:"p99_ms"`
+	ReadyRSSKB    int64   `json:"ready_rss_kb,omitempty"`
+	PostLoadRSSKB int64   `json:"post_load_rss_kb,omitempty"`
+	ArtifactBytes int64   `json:"artifact_bytes,omitempty"`
+}
+
+type benchmarkMetadata struct {
+	Type              string `json:"type"`
+	GOOS              string `json:"goos"`
+	GOARCH            string `json:"goarch"`
+	Rounds            int    `json:"rounds"`
+	StartupWarmups    int    `json:"startup_warmups"`
+	StartupIterations int    `json:"startup_iterations"`
+	WarmupRequests    int    `json:"warmup_requests"`
+	RPCRequests       int    `json:"rpc_requests"`
+	ManagementCalls   int    `json:"management_requests"`
+}
+
+func main() {
+	rounds := envInt("BENCH_ROUNDS", 5)
+	startupWarmups := envInt("BENCH_STARTUP_WARMUPS", 3)
+	startupIterations := envInt("BENCH_STARTUPS", 30)
+	warmupRequests := envInt("BENCH_WARMUP_REQUESTS", 500)
+	rpcRequests := envInt("BENCH_RPC_REQUESTS", 5000)
+	managementRequests := envInt("BENCH_MANAGEMENT_REQUESTS", 1000)
+
+	agents := []agentSpec{
+		{
+			Name:         "java",
+			Command:      javaAgentCommand(requiredEnv("JAVA_AGENT_JAR")),
+			ArtifactPath: requiredEnv("JAVA_AGENT_JAR"),
+		},
+		{
+			Name:         "go",
+			Command:      []string{requiredEnv("GO_AGENT")},
+			ArtifactPath: requiredEnv("GO_AGENT"),
+		},
+	}
+
+	encoder := json.NewEncoder(os.Stdout)
+	encode(encoder, benchmarkMetadata{
+		Type:              "metadata",
+		GOOS:              runtime.GOOS,
+		GOARCH:            runtime.GOARCH,
+		Rounds:            rounds,
+		StartupWarmups:    startupWarmups,
+		StartupIterations: startupIterations,
+		WarmupRequests:    warmupRequests,
+		RPCRequests:       rpcRequests,
+		ManagementCalls:   managementRequests,
+	})
+
+	for _, result := range benchmarkStartups(agents, startupWarmups, startupIterations) {
+		encode(encoder, result)
+	}
+
+	managementURL, closeManagementServer := startManagementServer()
+	defer closeManagementServer()
+	managementParams := map[string]any{
+		"connection": map[string]any{
+			"management_url": managementURL,
+			"username":       "guest",
+			"password":       "guest",
+			"virtual_host":   "/",
+		},
+	}
+
+	for round := 1; round <= rounds; round++ {
+		order := agents
+		if round%2 == 0 {
+			order = []agentSpec{agents[1], agents[0]}
+		}
+		for _, agent := range order {
+			encode(encoder, benchmarkRPC(agent, "handshake", round, "handshake", map[string]any{}, warmupRequests, rpcRequests))
+			encode(encoder, benchmarkRPC(
+				agent,
+				"management_list_topics",
+				round,
+				"mq_list_topics",
+				managementParams,
+				warmupRequests/5,
+				managementRequests,
+			))
+		}
+	}
+}
+
+func benchmarkStartups(agents []agentSpec, warmups, iterations int) []benchmarkResult {
+	for warmup := 0; warmup < warmups; warmup++ {
+		order := agents
+		if warmup%2 == 1 {
+			order = []agentSpec{agents[1], agents[0]}
+		}
+		for _, agent := range order {
+			process, _, err := startAgent(agent.Command)
+			if err != nil {
+				panic(fmt.Errorf("warm up startup %s: %w", agent.Name, err))
+			}
+			if _, err := process.call("handshake", map[string]any{}); err != nil {
+				process.kill()
+				panic(fmt.Errorf("warm up handshake %s: %w", agent.Name, err))
+			}
+			if err := process.close(); err != nil {
+				panic(fmt.Errorf("close startup warmup %s: %w", agent.Name, err))
+			}
+		}
+	}
+
+	readySamples := map[string][]float64{}
+	handshakeSamples := map[string][]float64{}
+	rssSamples := map[string][]int64{}
+	readyDurations := map[string]time.Duration{}
+	handshakeDurations := map[string]time.Duration{}
+	for iteration := 0; iteration < iterations; iteration++ {
+		order := agents
+		if iteration%2 == 1 {
+			order = []agentSpec{agents[1], agents[0]}
+		}
+		for _, agent := range order {
+			process, readyDuration, err := startAgent(agent.Command)
+			if err != nil {
+				panic(fmt.Errorf("start %s: %w", agent.Name, err))
+			}
+			handshakeStart := time.Now()
+			if _, err := process.call("handshake", map[string]any{}); err != nil {
+				process.kill()
+				panic(fmt.Errorf("handshake %s: %w", agent.Name, err))
+			}
+			handshakeDuration := time.Since(handshakeStart)
+			readySamples[agent.Name] = append(readySamples[agent.Name], milliseconds(readyDuration))
+			handshakeSamples[agent.Name] = append(
+				handshakeSamples[agent.Name],
+				milliseconds(readyDuration+handshakeDuration),
+			)
+			rssSamples[agent.Name] = append(rssSamples[agent.Name], readRSSKB(process.command.Process.Pid))
+			readyDurations[agent.Name] += readyDuration
+			handshakeDurations[agent.Name] += readyDuration + handshakeDuration
+			if err := process.close(); err != nil {
+				panic(fmt.Errorf("close %s: %w", agent.Name, err))
+			}
+		}
+	}
+
+	results := make([]benchmarkResult, 0, len(agents)*2)
+	for _, agent := range agents {
+		artifactBytes := fileSize(agent.ArtifactPath)
+		ready := summarize(agent.Name, "startup_ready", 0, readySamples[agent.Name], readyDurations[agent.Name], 0)
+		ready.ReadyRSSKB = medianInt64(rssSamples[agent.Name])
+		ready.ArtifactBytes = artifactBytes
+		results = append(results, ready)
+		withHandshake := summarize(
+			agent.Name,
+			"startup_handshake",
+			0,
+			handshakeSamples[agent.Name],
+			handshakeDurations[agent.Name],
+			0,
+		)
+		withHandshake.ReadyRSSKB = medianInt64(rssSamples[agent.Name])
+		withHandshake.ArtifactBytes = artifactBytes
+		results = append(results, withHandshake)
+	}
+	return results
+}
+
+func benchmarkRPC(
+	agent agentSpec,
+	workload string,
+	round int,
+	method string,
+	params map[string]any,
+	warmupRequests int,
+	operations int,
+) benchmarkResult {
+	process, _, err := startAgent(agent.Command)
+	if err != nil {
+		panic(fmt.Errorf("start %s: %w", agent.Name, err))
+	}
+	defer func() {
+		if err := process.close(); err != nil {
+			panic(fmt.Errorf("close %s: %w", agent.Name, err))
+		}
+	}()
+	readyRSS := readRSSKB(process.command.Process.Pid)
+	for request := 0; request < warmupRequests; request++ {
+		if _, err := process.call(method, params); err != nil {
+			panic(fmt.Errorf("warm up %s/%s: %w", agent.Name, workload, err))
+		}
+	}
+
+	latencies := make([]float64, 0, operations)
+	errorsCount := 0
+	start := time.Now()
+	for operation := 0; operation < operations; operation++ {
+		requestStart := time.Now()
+		if _, err := process.call(method, params); err != nil {
+			errorsCount++
+		}
+		latencies = append(latencies, milliseconds(time.Since(requestStart)))
+	}
+	duration := time.Since(start)
+	result := summarize(agent.Name, workload, round, latencies, duration, errorsCount)
+	result.ReadyRSSKB = readyRSS
+	result.PostLoadRSSKB = readRSSKB(process.command.Process.Pid)
+	result.ArtifactBytes = fileSize(agent.ArtifactPath)
+	return result
+}
+
+func summarize(
+	agent string,
+	workload string,
+	round int,
+	latencies []float64,
+	duration time.Duration,
+	errorsCount int,
+) benchmarkResult {
+	sorted := append([]float64(nil), latencies...)
+	sort.Float64s(sorted)
+	total := 0.0
+	for _, latency := range sorted {
+		total += latency
+	}
+	operations := len(sorted)
+	mean := 0.0
+	qps := 0.0
+	if operations > 0 {
+		mean = total / float64(operations)
+	}
+	if duration > 0 {
+		qps = float64(operations) / duration.Seconds()
+	}
+	return benchmarkResult{
+		Agent:      agent,
+		Workload:   workload,
+		Round:      round,
+		Operations: operations,
+		Errors:     errorsCount,
+		DurationMS: milliseconds(duration),
+		QPS:        qps,
+		MeanMS:     mean,
+		P50MS:      percentile(sorted, 0.50),
+		P95MS:      percentile(sorted, 0.95),
+		P99MS:      percentile(sorted, 0.99),
+	}
+}
+
+func startAgent(command []string) (*agentProcess, time.Duration, error) {
+	if len(command) == 0 {
+		return nil, 0, errors.New("empty agent command")
+	}
+	process := &agentProcess{}
+	process.command = exec.Command(command[0], command[1:]...)
+	process.command.Env = sanitizedEnv()
+	stdin, err := process.command.StdinPipe()
+	if err != nil {
+		return nil, 0, err
+	}
+	stdout, err := process.command.StdoutPipe()
+	if err != nil {
+		return nil, 0, err
+	}
+	process.command.Stderr = os.Stderr
+	process.stdin = stdin
+	process.reader = bufio.NewScanner(stdout)
+	process.reader.Buffer(make([]byte, 64*1024), 512*1024*1024)
+	start := time.Now()
+	if err := process.command.Start(); err != nil {
+		return nil, 0, err
+	}
+	if !process.reader.Scan() {
+		process.kill()
+		return nil, 0, fmt.Errorf("agent did not become ready: %v", process.reader.Err())
+	}
+	if !strings.Contains(process.reader.Text(), `"ready":true`) {
+		process.kill()
+		return nil, 0, fmt.Errorf("agent did not become ready: %s", process.reader.Text())
+	}
+	return process, time.Since(start), nil
+}
+
+func (process *agentProcess) call(method string, params map[string]any) (json.RawMessage, error) {
+	process.nextID++
+	request := map[string]any{
+		"jsonrpc": "2.0",
+		"id":      process.nextID,
+		"method":  method,
+		"params":  params,
+	}
+	payload, err := json.Marshal(request)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := process.stdin.Write(append(payload, '\n')); err != nil {
+		return nil, err
+	}
+	if !process.reader.Scan() {
+		return nil, fmt.Errorf("agent response unavailable: %v", process.reader.Err())
+	}
+	var response agentResponse
+	if err := json.Unmarshal(process.reader.Bytes(), &response); err != nil {
+		return nil, err
+	}
+	if response.ID != process.nextID {
+		return nil, fmt.Errorf("response id %d does not match request id %d", response.ID, process.nextID)
+	}
+	if response.Error != nil {
+		return nil, errors.New(response.Error.Message)
+	}
+	return response.Result, nil
+}
+
+func (process *agentProcess) close() error {
+	_, callError := process.call("shutdown", map[string]any{})
+	_ = process.stdin.Close()
+	waitError := process.command.Wait()
+	if callError != nil {
+		return callError
+	}
+	return waitError
+}
+
+func (process *agentProcess) kill() {
+	if process != nil && process.command != nil && process.command.Process != nil {
+		_ = process.command.Process.Kill()
+		_, _ = process.command.Process.Wait()
+	}
+}
+
+func startManagementServer() (string, func()) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		panic(err)
+	}
+	queues := make([]map[string]any, 0, 12)
+	for index := 11; index >= 0; index-- {
+		queues = append(queues, map[string]any{
+			"name":        fmt.Sprintf("queue-%02d", index),
+			"durable":     index%2 == 0,
+			"auto_delete": index%3 == 0,
+			"state":       "running",
+			"messages":    index * 100,
+			"consumers":   index % 4,
+		})
+	}
+	body, err := json.Marshal(map[string]any{
+		"items":       queues,
+		"page":        1,
+		"page_count":  1,
+		"total_count": len(queues),
+	})
+	if err != nil {
+		panic(err)
+	}
+	server := &http.Server{Handler: http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = writer.Write(body)
+	})}
+	go func() {
+		if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			panic(err)
+		}
+	}()
+	return "http://" + listener.Addr().String(), func() {
+		_ = server.Close()
+	}
+}
+
+func javaAgentCommand(jarPath string) []string {
+	return []string{
+		"java",
+		"-Dfile.encoding=UTF-8",
+		"-Dsun.stdout.encoding=UTF-8",
+		"-Dsun.stderr.encoding=UTF-8",
+		"-Djava.net.useSystemProxies=false",
+		"-Dhttp.proxyHost=",
+		"-Dhttps.proxyHost=",
+		"-DsocksProxyHost=",
+		"-Doracle.net.disableOob=true",
+		"-Doracle.jdbc.javaNetNio=false",
+		"--add-opens=java.sql/java.sql=ALL-UNNAMED",
+		"-XX:TieredStopAtLevel=1",
+		"-XX:+UseSerialGC",
+		"-jar",
+		jarPath,
+	}
+}
+
+func sanitizedEnv() []string {
+	blocked := map[string]struct{}{
+		"HTTP_PROXY": {}, "HTTPS_PROXY": {}, "ALL_PROXY": {}, "NO_PROXY": {},
+		"http_proxy": {}, "https_proxy": {}, "all_proxy": {}, "no_proxy": {},
+	}
+	result := make([]string, 0, len(os.Environ()))
+	for _, variable := range os.Environ() {
+		key, _, _ := strings.Cut(variable, "=")
+		if _, skip := blocked[key]; !skip {
+			result = append(result, variable)
+		}
+	}
+	return result
+}
+
+func percentile(sorted []float64, ratio float64) float64 {
+	if len(sorted) == 0 {
+		return 0
+	}
+	index := int(math.Ceil(ratio*float64(len(sorted)))) - 1
+	if index < 0 {
+		index = 0
+	}
+	return sorted[index]
+}
+
+func medianInt64(values []int64) int64 {
+	if len(values) == 0 {
+		return 0
+	}
+	sorted := append([]int64(nil), values...)
+	sort.Slice(sorted, func(left, right int) bool { return sorted[left] < sorted[right] })
+	return sorted[len(sorted)/2]
+}
+
+func readRSSKB(processID int) int64 {
+	output, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(processID)).Output()
+	if err != nil {
+		return 0
+	}
+	value, err := strconv.ParseInt(strings.TrimSpace(string(output)), 10, 64)
+	if err != nil {
+		return 0
+	}
+	return value
+}
+
+func fileSize(path string) int64 {
+	info, err := os.Stat(path)
+	if err != nil {
+		panic(err)
+	}
+	return info.Size()
+}
+
+func milliseconds(duration time.Duration) float64 {
+	return float64(duration.Nanoseconds()) / float64(time.Millisecond)
+}
+
+func requiredEnv(key string) string {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		panic(key + " is required")
+	}
+	return value
+}
+
+func envInt(key string, fallback int) int {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil || parsed < 1 {
+		panic(key + " must be a positive integer")
+	}
+	return parsed
+}
+
+func encode(encoder *json.Encoder, value any) {
+	if err := encoder.Encode(value); err != nil {
+		panic(err)
+	}
+}

agents/drivers/rabbitmq/build.gradle

@@ -1,12 +0,0 @@
-dependencies {
-    implementation 'com.google.code.gson:gson:2.12.1'
-    implementation 'com.rabbitmq:amqp-client:5.21.0'
-    runtimeOnly 'org.slf4j:slf4j-simple:1.7.36'
-}
-
-tasks.named('shadowJar') {
-    mergeServiceFiles()
-    manifest {
-        attributes('Agent-Label': 'RabbitMQ', 'Main-Class': 'com.dbx.agent.rabbitmq.RabbitMqAgent')
-    }
-}

agents/drivers/rabbitmq/go.mod

@@ -0,0 +1,5 @@
+module github.com/t8y2/dbx/agents/drivers/rabbitmq
+
+go 1.22
+
+require github.com/rabbitmq/amqp091-go v1.13.0

agents/drivers/rabbitmq/go.sum

@@ -0,0 +1,4 @@
+github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
+github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=

agents/drivers/rabbitmq/helpers.go

@@ -0,0 +1,448 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+var (
+	quotedNamePattern       = regexp.MustCompile(`'([^']+)'`)
+	declaredResourcePattern = regexp.MustCompile(`for (queue|exchange) '([^']+)'`)
+)
+
+func decodeJSON(data []byte, target any) error {
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	decoder.UseNumber()
+	return decoder.Decode(target)
+}
+
+func deepCopyObject(source jsonObject) jsonObject {
+	if source == nil {
+		return nil
+	}
+	encoded, err := json.Marshal(source)
+	if err != nil {
+		return nil
+	}
+	copy := jsonObject{}
+	if err := decodeJSON(encoded, &copy); err != nil {
+		return nil
+	}
+	return copy
+}
+
+func okResult() jsonObject {
+	return jsonObject{"ok": true}
+}
+
+func objectOrNil(object jsonObject, key string) jsonObject {
+	if object == nil {
+		return nil
+	}
+	switch value := object[key].(type) {
+	case jsonObject:
+		return value
+	case map[string]any:
+		return jsonObject(value)
+	default:
+		return nil
+	}
+}
+
+func arrayOrNil(object jsonObject, key string) []any {
+	if object == nil {
+		return nil
+	}
+	array, _ := object[key].([]any)
+	return array
+}
+
+func stringOrNull(object jsonObject, key string) *string {
+	if object == nil {
+		return nil
+	}
+	value, exists := object[key]
+	if !exists || value == nil {
+		return nil
+	}
+	var result string
+	switch typed := value.(type) {
+	case string:
+		result = typed
+	case json.Number:
+		result = typed.String()
+	case bool:
+		result = strconv.FormatBool(typed)
+	case float64:
+		result = strconv.FormatFloat(typed, 'f', -1, 64)
+	default:
+		result = fmt.Sprint(typed)
+	}
+	return &result
+}
+
+func stringOrEmpty(object jsonObject, key string) string {
+	return stringOrDefault(object, key, "")
+}
+
+func stringOrDefault(object jsonObject, key, fallback string) string {
+	value := stringOrNull(object, key)
+	if value == nil {
+		return fallback
+	}
+	return *value
+}
+
+func integerOrNull(object jsonObject, key string) *int {
+	value, ok := numberAsInt64(object, key)
+	if !ok {
+		return nil
+	}
+	converted := int(value)
+	return &converted
+}
+
+func longOrNull(object jsonObject, key string) *int64 {
+	value, ok := numberAsInt64(object, key)
+	if !ok {
+		return nil
+	}
+	return &value
+}
+
+func numberAsInt64(object jsonObject, key string) (int64, bool) {
+	if object == nil {
+		return 0, false
+	}
+	value, exists := object[key]
+	if !exists || value == nil {
+		return 0, false
+	}
+	switch typed := value.(type) {
+	case json.Number:
+		if integer, err := typed.Int64(); err == nil {
+			return integer, true
+		}
+		decimal, err := typed.Float64()
+		return int64(decimal), err == nil
+	case float64:
+		return int64(typed), true
+	case float32:
+		return int64(typed), true
+	case int:
+		return int64(typed), true
+	case int8:
+		return int64(typed), true
+	case int16:
+		return int64(typed), true
+	case int32:
+		return int64(typed), true
+	case int64:
+		return typed, true
+	case uint:
+		return int64(typed), true
+	case uint8:
+		return int64(typed), true
+	case uint16:
+		return int64(typed), true
+	case uint32:
+		return int64(typed), true
+	case uint64:
+		return int64(typed), true
+	case string:
+		integer, err := strconv.ParseInt(typed, 10, 64)
+		return integer, err == nil
+	default:
+		return 0, false
+	}
+}
+
+func intOrDefault(object jsonObject, key string, fallback int) int {
+	value := integerOrNull(object, key)
+	if value == nil {
+		return fallback
+	}
+	return *value
+}
+
+func longOrDefault(object jsonObject, key string, fallback int64) int64 {
+	value := longOrNull(object, key)
+	if value == nil {
+		return fallback
+	}
+	return *value
+}
+
+func floatOrNull(object jsonObject, key string) *float64 {
+	if object == nil {
+		return nil
+	}
+	value, exists := object[key]
+	if !exists || value == nil {
+		return nil
+	}
+	var result float64
+	var err error
+	switch typed := value.(type) {
+	case json.Number:
+		result, err = typed.Float64()
+	case float64:
+		result = typed
+	case float32:
+		result = float64(typed)
+	case int:
+		result = float64(typed)
+	case int64:
+		result = float64(typed)
+	case string:
+		result, err = strconv.ParseFloat(typed, 64)
+	default:
+		return nil
+	}
+	if err != nil {
+		return nil
+	}
+	return &result
+}
+
+func boolOrDefault(object jsonObject, key string, fallback bool) bool {
+	if object == nil {
+		return fallback
+	}
+	value, exists := object[key]
+	if !exists || value == nil {
+		return fallback
+	}
+	switch typed := value.(type) {
+	case bool:
+		return typed
+	case string:
+		parsed, err := strconv.ParseBool(typed)
+		if err == nil {
+			return parsed
+		}
+	}
+	return fallback
+}
+
+func integerProperty(properties jsonObject, key string) (int, bool) {
+	if properties == nil {
+		return 0, false
+	}
+	value := integerOrNull(properties, key)
+	if value == nil {
+		return 0, false
+	}
+	return *value, true
+}
+
+func boolProperty(config jsonObject, key string) bool {
+	return boolOrDefault(objectOrNil(config, "properties"), key, false)
+}
+
+func durationMilliseconds(object jsonObject, key string, fallback time.Duration) time.Duration {
+	value := integerOrNull(object, key)
+	if value == nil {
+		return fallback
+	}
+	return time.Duration(*value) * time.Millisecond
+}
+
+func argumentValue(value any) any {
+	switch typed := value.(type) {
+	case nil:
+		return nil
+	case bool, string:
+		return typed
+	case json.Number:
+		if integer, err := typed.Int64(); err == nil {
+			return integer
+		}
+		if decimal, err := typed.Float64(); err == nil {
+			return int64(decimal)
+		}
+		return nil
+	case float64:
+		return int64(typed)
+	case float32:
+		return int64(typed)
+	case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
+		return typed
+	default:
+		return nil
+	}
+}
+
+func connectionObject(params jsonObject) jsonObject {
+	if connection := objectOrNil(params, "connection"); connection != nil {
+		return connection
+	}
+	return params
+}
+
+func (s *server) currentConnectionConfig(params jsonObject) jsonObject {
+	if connection := objectOrNil(params, "connection"); connection != nil {
+		return connection
+	}
+	return s.cachedConnection
+}
+
+func (s *server) requireConnectionConfig(params jsonObject) (jsonObject, error) {
+	connection := s.currentConnectionConfig(params)
+	if connection == nil {
+		return nil, errors.New("Not connected. Call connect first.")
+	}
+	return connection, nil
+}
+
+func (s *server) requireConnection() (*amqp.Connection, error) {
+	if s.connection == nil {
+		return nil, errors.New("Not connected. Call connect first.")
+	}
+	return s.connection, nil
+}
+
+func queueName(params jsonObject) (string, error) {
+	name := stringOrEmpty(params, "topic")
+	if strings.TrimSpace(name) == "" {
+		name = stringOrEmpty(params, "name")
+	}
+	if strings.TrimSpace(name) == "" {
+		return "", errors.New("topic (queue name) is required")
+	}
+	return name, nil
+}
+
+func effectiveVhost(params, connection jsonObject) string {
+	vhost := stringOrNull(params, "virtual_host")
+	if vhost == nil || strings.TrimSpace(*vhost) == "" {
+		if connection != nil {
+			return stringOrDefault(connection, "virtual_host", "/")
+		}
+		return "/"
+	}
+	return *vhost
+}
+
+func allVhostsRequested(params jsonObject) bool {
+	return boolOrDefault(params, "all_vhosts", false)
+}
+
+func managementListPath(params, connection jsonObject, resource string) string {
+	if allVhostsRequested(params) {
+		return "/api/" + resource
+	}
+	return "/api/" + resource + "/" + urlEncodeVhost(effectiveVhost(params, connection))
+}
+
+func vhostFilter(params, connection jsonObject) string {
+	if allVhostsRequested(params) {
+		return ""
+	}
+	return effectiveVhost(params, connection)
+}
+
+func attachVhost(info jsonObject, source jsonObject) {
+	info["vhost"] = stringOrEmpty(source, "vhost")
+}
+
+func serverString(properties amqp.Table, key string) any {
+	value, exists := properties[key]
+	if !exists || value == nil {
+		return nil
+	}
+	switch typed := value.(type) {
+	case []byte:
+		return string(typed)
+	default:
+		return fmt.Sprint(typed)
+	}
+}
+
+func normalizeErrorMessage(err error) string {
+	if err == nil {
+		return "error"
+	}
+	if errors.Is(err, amqp.ErrCredentials) {
+		return err.Error() + ". Hint: authentication failed. Check the RabbitMQ username, password, and virtual host permissions."
+	}
+	var amqpError *amqp.Error
+	if errors.As(err, &amqpError) {
+		if friendly := mapAMQPError(amqpError.Code, amqpError.Reason); friendly != "" {
+			return friendly
+		}
+	}
+	message := strings.TrimSpace(err.Error())
+	if message == "" {
+		return fmt.Sprintf("%T", err)
+	}
+	return message
+}
+
+func mapAMQPError(replyCode int, replyText string) string {
+	switch replyCode {
+	case 405:
+		name := extractQuotedName(replyText)
+		subject := "The queue"
+		if name != "" {
+			subject = "Queue '" + name + "'"
+		}
+		return subject + " is exclusive and owned by another connection. Hint: exclusive queues can only be accessed by their owning connection; stats via the management API are still available."
+	case 404:
+		name := extractQuotedName(replyText)
+		kind := "Queue"
+		if strings.Contains(replyText, "no exchange") {
+			kind = "Exchange"
+		}
+		subject := "The " + strings.ToLower(kind)
+		if name != "" {
+			subject = kind + " '" + name + "'"
+		}
+		return subject + " was not found. Hint: it may have been deleted, or it never existed on this virtual host."
+	case 406:
+		name := extractDeclaredResourceName(replyText)
+		kind := "Queue"
+		if strings.Contains(replyText, "for exchange") {
+			kind = "Exchange"
+		}
+		subject := "The " + strings.ToLower(kind)
+		if name != "" {
+			subject = kind + " '" + name + "'"
+		}
+		lowerKind := strings.ToLower(kind)
+		return subject + " already exists with different parameters. Hint: " + lowerKind + " parameters are immutable after declaration; delete and re-declare the " + lowerKind + " to change them."
+	case 403:
+		name := extractQuotedName(replyText)
+		subject := "the requested resource"
+		if name != "" {
+			subject = "'" + name + "'"
+		}
+		return "Access to " + subject + " was refused. Hint: check the user's configure/write/read permissions on the virtual host."
+	default:
+		return ""
+	}
+}
+
+func extractQuotedName(replyText string) string {
+	match := quotedNamePattern.FindStringSubmatch(replyText)
+	if len(match) < 2 {
+		return ""
+	}
+	return match[1]
+}
+
+func extractDeclaredResourceName(replyText string) string {
+	match := declaredResourcePattern.FindStringSubmatch(replyText)
+	if len(match) < 3 {
+		return ""
+	}
+	return match[2]
+}

agents/drivers/rabbitmq/helpers_test.go

@@ -0,0 +1,305 @@
+package main
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func mustObject(t *testing.T, source string) jsonObject {
+	t.Helper()
+	result := jsonObject{}
+	if err := decodeJSON([]byte(source), &result); err != nil {
+		t.Fatal(err)
+	}
+	return result
+}
+
+func TestParseAddresses(t *testing.T) {
+	tests := []struct {
+		name        string
+		value       string
+		defaultPort int
+		want        []address
+		wantError   string
+	}{
+		{name: "pairs", value: "host1:5672,host2:5673", defaultPort: 5672, want: []address{{"host1", 5672}, {"host2", 5673}}},
+		{name: "bare host", value: "rabbit", defaultPort: 5679, want: []address{{"rabbit", 5679}}},
+		{name: "blank entries", value: " , rabbit:5672, ", defaultPort: 5679, want: []address{{"rabbit", 5672}}},
+		{name: "ipv6", value: "[::1]:5672", defaultPort: 5679, want: []address{{"::1", 5672}}},
+		{name: "blank", value: " , ", defaultPort: 5672, wantError: "addresses is required"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := parseAddresses(test.value, test.defaultPort)
+			if test.wantError != "" {
+				if err == nil || !strings.Contains(err.Error(), test.wantError) {
+					t.Fatalf("got error %v, want %q", err, test.wantError)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(got) != len(test.want) {
+				t.Fatalf("got %#v, want %#v", got, test.want)
+			}
+			for index := range got {
+				if got[index] != test.want[index] {
+					t.Fatalf("got %#v, want %#v", got, test.want)
+				}
+			}
+		})
+	}
+}
+
+func TestResolveAddresses(t *testing.T) {
+	tests := []struct {
+		name      string
+		config    jsonObject
+		want      []address
+		wantError string
+	}{
+		{name: "explicit port", config: mustObject(t, `{"addresses":"rabbit","port":5679}`), want: []address{{"rabbit", 5679}}},
+		{name: "default port", config: mustObject(t, `{"addresses":"rabbit"}`), want: []address{{"rabbit", 5672}}},
+		{name: "host fallback", config: mustObject(t, `{"host":"rabbit"}`), want: []address{{"rabbit", 5672}}},
+		{name: "missing", config: mustObject(t, `{}`), wantError: "addresses is required"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := resolveAddresses(test.config)
+			if test.wantError != "" {
+				if err == nil || !strings.Contains(err.Error(), test.wantError) {
+					t.Fatalf("got error %v, want %q", err, test.wantError)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(got) != len(test.want) || got[0] != test.want[0] {
+				t.Fatalf("got %#v, want %#v", got, test.want)
+			}
+		})
+	}
+}
+
+func TestPeekNormalizationAndRoutingKey(t *testing.T) {
+	if normalizePeekOffset(-1) != 0 || normalizePeekOffset(4) != 4 {
+		t.Fatal("unexpected offset normalization")
+	}
+	if normalizePeekCount(0) != 1 || normalizePeekCount(8) != 8 {
+		t.Fatal("unexpected count normalization")
+	}
+	tests := []struct {
+		params jsonObject
+		want   string
+	}{
+		{mustObject(t, `{"routing_key":"explicit","routingKey":"camel","key":"message"}`), "explicit"},
+		{mustObject(t, `{"routingKey":"camel","key":"message"}`), "camel"},
+		{mustObject(t, `{"key":"message"}`), "message"},
+		{mustObject(t, `{"key":"  "}`), "queue"},
+		{mustObject(t, `{}`), "queue"},
+	}
+	for _, test := range tests {
+		if got := resolveRoutingKey(test.params, "queue"); got != test.want {
+			t.Fatalf("got %q, want %q", got, test.want)
+		}
+	}
+	if got := peekMessageCapacity(10, 3, int(^uint(0)>>1)); got != 7 {
+		t.Fatalf("unexpected bounded capacity %d", got)
+	}
+	if got := peekMessageCapacity(2, 5, 10); got != 0 {
+		t.Fatalf("unexpected exhausted capacity %d", got)
+	}
+}
+
+func TestTLSAndManagementConfiguration(t *testing.T) {
+	if tlsSkipVerify(mustObject(t, `{"tls_skip_verify":true}`)) != true {
+		t.Fatal("top-level skip verify not detected")
+	}
+	if tlsSkipVerify(mustObject(t, `{"tls":{"skip_verify":true}}`)) != true {
+		t.Fatal("nested skip verify not detected")
+	}
+	if managementTLS(mustObject(t, `{"tls_skip_verify":true}`)) {
+		t.Fatal("skip verify must not enable management TLS")
+	}
+	if !managementTLS(mustObject(t, `{"tls":{}}`)) || !managementTLS(mustObject(t, `{"properties":{"ssl":true}}`)) {
+		t.Fatal("management TLS not detected")
+	}
+	if managementPort(jsonObject{}, false) != 15672 || managementPort(jsonObject{}, true) != 15671 {
+		t.Fatal("unexpected default management ports")
+	}
+	if managementPort(mustObject(t, `{"properties":{"management_port":55672}}`), false) != 55672 {
+		t.Fatal("management port override ignored")
+	}
+}
+
+func TestCredentialAndAuthHelpers(t *testing.T) {
+	config := mustObject(t, `{"username":" ","password":null}`)
+	if credentialOrGuest(config, "username") != "guest" || credentialOrGuest(config, "password") != "guest" {
+		t.Fatal("blank credentials did not fall back to guest")
+	}
+	want := "Basic " + base64.StdEncoding.EncodeToString([]byte("guest:guest"))
+	if got := basicAuthHeader("guest", "guest"); got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}
+
+func TestPathEncoding(t *testing.T) {
+	if got := urlEncodeVhost("/"); got != "%2F" {
+		t.Fatalf("got %q", got)
+	}
+	if got := urlEncodePathSegment("queue one"); got != "queue%20one" {
+		t.Fatalf("got %q", got)
+	}
+	if got := urlEncodeName("127.0.0.1:1 -> 127.0.0.1:2"); !strings.Contains(got, "%20-%3E%20") {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestHandshakeAndRequestErrors(t *testing.T) {
+	service := newServer()
+	response, shutdown := service.handleRequest([]byte(`{"jsonrpc":"2.0","id":1,"method":"handshake","params":{}}`))
+	if shutdown || response.Error != nil {
+		t.Fatalf("unexpected response: %#v", response)
+	}
+	result, ok := response.Result.(handshakeResult)
+	if !ok || result.ProtocolVersion != 1 || result.AgentProtocolVersion != 1 || len(result.Capabilities) != len(capabilities) {
+		t.Fatalf("unexpected handshake: %#v", response.Result)
+	}
+	response, _ = service.handleRequest([]byte(`{"jsonrpc":"2.0","id":2,"method":"unknown","params":{}}`))
+	if response.Error == nil || !strings.Contains(response.Error.Message, "Unknown method") {
+		t.Fatalf("unexpected response: %#v", response)
+	}
+	response, _ = service.handleRequest([]byte(`not json`))
+	if response.Error == nil || string(response.ID) != "null" {
+		t.Fatalf("unexpected malformed response: %#v", response)
+	}
+	response, _ = service.handleRequest([]byte(`{"jsonrpc":"2.0","id":7,"params":{}}`))
+	if response.Error == nil || string(response.ID) != "7" {
+		t.Fatalf("unexpected missing-method response: %#v", response)
+	}
+	encoded, err := json.Marshal(response)
+	if err != nil || !strings.Contains(string(encoded), `"id":7`) {
+		t.Fatalf("unexpected JSON: %s, %v", encoded, err)
+	}
+}
+
+func TestAllVhostsGuardsAndEffectiveVhost(t *testing.T) {
+	service := newServer()
+	for method := range allVhostsUnsupportedMethods {
+		_, _, err := service.dispatch(method, mustObject(t, `{"all_vhosts":true}`))
+		if err == nil || err.Error() != "all_vhosts is only supported for list operations" {
+			t.Fatalf("%s: %v", method, err)
+		}
+	}
+	connection := mustObject(t, `{"virtual_host":"connected"}`)
+	if got := effectiveVhost(mustObject(t, `{"virtual_host":"explicit"}`), connection); got != "explicit" {
+		t.Fatalf("got %q", got)
+	}
+	if got := effectiveVhost(mustObject(t, `{"virtual_host":" "}`), connection); got != "connected" {
+		t.Fatalf("got %q", got)
+	}
+	if got := effectiveVhost(jsonObject{}, nil); got != "/" {
+		t.Fatalf("got %q", got)
+	}
+	if allVhostsRequested(jsonObject{}) {
+		t.Fatal("all_vhosts should default false")
+	}
+	if got := managementListPath(jsonObject{}, connection, "queues"); got != "/api/queues/connected" {
+		t.Fatalf("got %q", got)
+	}
+	if got := managementListPath(mustObject(t, `{"all_vhosts":true,"virtual_host":"ignored"}`), connection, "queues"); got != "/api/queues" {
+		t.Fatalf("got %q", got)
+	}
+	if got := vhostFilter(mustObject(t, `{"all_vhosts":true}`), connection); got != "" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestSemanticGuards(t *testing.T) {
+	if _, err := queueName(jsonObject{}); err == nil || !strings.Contains(err.Error(), "queue name") {
+		t.Fatal(err)
+	}
+	if _, err := namespaceName(jsonObject{}); err == nil || err.Error() != "namespace is required" {
+		t.Fatal(err)
+	}
+	if _, err := namespaceName(mustObject(t, `{"namespace":"*"}`)); err == nil {
+		t.Fatal("all-vhosts namespace accepted")
+	}
+	if err := assertNamespaceDeletable("/", ""); err == nil {
+		t.Fatal("default vhost deletion accepted")
+	}
+	if err := assertNamespaceDeletable("orders", "orders"); err == nil {
+		t.Fatal("connected vhost deletion accepted")
+	}
+	for _, exchangeType := range []string{"direct", "fanout", "topic", "headers"} {
+		if _, err := validateExchangeType(exchangeType); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := validateExchangeType("stream"); err == nil {
+		t.Fatal("invalid exchange type accepted")
+	}
+	for _, name := range []string{"", "amq.direct"} {
+		if err := assertExchangeDeletable(name); err == nil {
+			t.Fatalf("exchange %q accepted", name)
+		}
+	}
+	if _, err := permissionVhost(jsonObject{}); err == nil {
+		t.Fatal("blank permission vhost accepted")
+	}
+	if _, err := permissionVhost(mustObject(t, `{"virtual_host":"*"}`)); err == nil {
+		t.Fatal("all-vhosts permission accepted")
+	}
+	if err := assertNotConnectedUser("delete", "dbx", "dbx"); err == nil {
+		t.Fatal("connected user mutation accepted")
+	}
+}
+
+func TestPermissionAndUserHelpers(t *testing.T) {
+	if permissionPattern(jsonObject{}, "read") != ".*" || permissionPattern(mustObject(t, `{"read":"^q"}`), "read") != "^q" {
+		t.Fatal("unexpected permission pattern")
+	}
+	if got := parseUserTags("administrator, management, ,policymaker"); len(got) != 3 || got[1] != "management" {
+		t.Fatalf("got %#v", got)
+	}
+	if got := userTagsParam(mustObject(t, `{"tags":["management"," policymaker ",""]}`)); got != "management,policymaker" {
+		t.Fatalf("got %q", got)
+	}
+	if got := userTagsParam(mustObject(t, `{"tags":"administrator,management"}`)); got != "administrator,management" {
+		t.Fatalf("got %q", got)
+	}
+}
+
+func TestAMQPErrorMapping(t *testing.T) {
+	tests := []struct {
+		code int
+		text string
+		want string
+	}{
+		{405, "RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'q1'", "Queue 'q1' is exclusive"},
+		{405, "RESOURCE_LOCKED", "The queue is exclusive"},
+		{404, "NOT_FOUND - no queue 'q1' in vhost '/'", "Queue 'q1' was not found"},
+		{404, "NOT_FOUND - no exchange 'events' in vhost '/'", "Exchange 'events' was not found"},
+		{406, "PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'q1' in vhost '/'", "Queue 'q1' already exists"},
+		{406, "PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'events' in vhost '/'", "Exchange 'events' already exists"},
+		{403, "ACCESS_REFUSED - access to queue 'q1' refused", "Access to 'q1' was refused"},
+	}
+	for _, test := range tests {
+		if got := mapAMQPError(test.code, test.text); !strings.Contains(got, test.want) {
+			t.Fatalf("got %q, want substring %q", got, test.want)
+		}
+	}
+	if got := mapAMQPError(320, "CONNECTION_FORCED"); got != "" {
+		t.Fatalf("unexpected mapping %q", got)
+	}
+	if got := extractDeclaredResourceName("inequivalent arg 'durable' for queue 'q1' in vhost '/'"); got != "q1" {
+		t.Fatalf("got %q", got)
+	}
+	if got := extractQuotedName("access to queue 'q1' refused for user 'dbx'"); got != "q1" {
+		t.Fatalf("got %q", got)
+	}
+}

agents/drivers/rabbitmq/integration_test.go

@@ -0,0 +1,248 @@
+package main
+
+import (
+	"encoding/base64"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestRabbitMQIntegration(t *testing.T) {
+	if os.Getenv("RABBITMQ_INTEGRATION") != "1" {
+		t.Skip("set RABBITMQ_INTEGRATION=1 to run against a real RabbitMQ broker")
+	}
+	host := envOrDefault("RABBITMQ_HOST", "127.0.0.1")
+	amqpPort := envIntOrDefault(t, "RABBITMQ_PORT", 5672)
+	managementPort := envIntOrDefault(t, "RABBITMQ_MANAGEMENT_PORT", 15672)
+	username := envOrDefault("RABBITMQ_USERNAME", "dbx")
+	password := envOrDefault("RABBITMQ_PASSWORD", "dbx-password")
+	connection := jsonObject{
+		"addresses": host,
+		"port":      amqpPort,
+		"username":  username,
+		"password":  password,
+		"properties": jsonObject{
+			"management_port": managementPort,
+		},
+	}
+	service := newServer()
+	if _, err := service.connect(connection); err != nil {
+		t.Fatal(err)
+	}
+	defer service.closeClients()
+
+	probe, err := service.testConnection(connection)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if probe.(jsonObject)["ok"] != true || probe.(jsonObject)["serverVersion"] == nil {
+		t.Fatalf("unexpected probe %#v", probe)
+	}
+	badConnection := deepCopyObject(connection)
+	badConnection["password"] = "definitely-wrong-password"
+	if _, err := service.testConnection(badConnection); err == nil {
+		t.Fatal("expected invalid credentials to fail")
+	} else if !strings.Contains(normalizeErrorMessage(err), "authentication failed") {
+		t.Fatalf("authentication error lost its actionable hint: %v", err)
+	}
+
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	vhost := "dbx-go-" + suffix
+	queue := "queue-" + suffix
+	exchange := "exchange-" + suffix
+	policy := "policy-" + suffix
+	user := "user-" + suffix
+
+	defer managementSend(connection, "DELETE", "/api/users/"+urlEncodePathSegment(user), nil)
+	defer managementSend(connection, "DELETE", "/api/vhosts/"+urlEncodeVhost(vhost), nil)
+
+	if _, err := service.createNamespace(jsonObject{"namespace": vhost}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.grantPermission(jsonObject{
+		"user": username, "virtual_host": vhost, "configure": ".*", "write": ".*", "read": ".*",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.getTopicStats(jsonObject{"topic": "missing-" + suffix, "virtual_host": vhost}); err == nil {
+		t.Fatal("expected missing queue lookup to fail")
+	} else if !strings.Contains(normalizeErrorMessage(err), "was not found") {
+		t.Fatalf("unexpected missing queue error: %v", err)
+	}
+	if _, err := service.createTopic(jsonObject{"topic": queue, "virtual_host": vhost, "durable": true}); err != nil {
+		t.Fatalf("channel did not recover after a broker-forced close: %v", err)
+	}
+	if _, err := service.createExchange(jsonObject{
+		"name": exchange, "type": "topic", "virtual_host": vhost, "durable": true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.bind(jsonObject{
+		"source": exchange, "destination": queue, "destinationType": "queue", "routingKey": "orders.*", "virtual_host": vhost,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	payload := "RabbitMQ Go Agent 世界"
+	if _, err := service.sendMessage(jsonObject{
+		"topic": queue, "exchange": exchange, "routingKey": "orders.created", "virtual_host": vhost,
+		"payloadBase64": base64.StdEncoding.EncodeToString([]byte(payload)),
+		"headers":       jsonObject{"source": "integration", "attempt": 1},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	peeked, err := service.peekMessages(jsonObject{"topic": queue, "virtual_host": vhost, "offset": 0, "count": 10})
+	if err != nil {
+		t.Fatal(err)
+	}
+	messages := peeked.(jsonObject)["messages"].([]jsonObject)
+	if len(messages) != 1 || messages[0]["payloadText"] != payload || messages[0]["routingKey"] != "orders.created" {
+		t.Fatalf("unexpected messages %#v", messages)
+	}
+
+	stats, err := service.getTopicStats(jsonObject{"topic": queue, "virtual_host": vhost})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.(jsonObject)["totalMessages"] != int64(1) {
+		t.Fatalf("unexpected stats %#v", stats)
+	}
+	config, err := service.getTopicConfig(jsonObject{"topic": queue, "virtual_host": vhost})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if config.(jsonObject)["configs"].(jsonObject)["durable"] != true {
+		t.Fatalf("unexpected config %#v", config)
+	}
+	consumers, err := service.listConsumers(jsonObject{"topic": queue, "virtual_host": vhost})
+	if err != nil || len(consumers.(jsonObject)["consumers"].([]jsonObject)) != 0 {
+		t.Fatalf("unexpected consumers %#v, %v", consumers, err)
+	}
+
+	topics, err := service.listTopics(jsonObject{"virtual_host": vhost})
+	if err != nil || !containsNamedItem(topics.(jsonObject)["topics"].([]jsonObject), queue) {
+		t.Fatalf("unexpected topics %#v, %v", topics, err)
+	}
+	exchanges, err := service.listExchanges(jsonObject{"virtual_host": vhost})
+	if err != nil || !containsNamedItem(exchanges.(jsonObject)["exchanges"].([]jsonObject), exchange) {
+		t.Fatalf("unexpected exchanges %#v, %v", exchanges, err)
+	}
+	bindings, err := service.listBindings(jsonObject{"virtual_host": vhost, "queue": queue})
+	if err != nil || len(bindings.(jsonObject)["bindings"].([]jsonObject)) == 0 {
+		t.Fatalf("unexpected bindings %#v, %v", bindings, err)
+	}
+
+	if _, err := service.setPolicy(jsonObject{
+		"virtual_host": vhost, "name": policy, "pattern": "^" + queue + "$", "applyTo": "queues",
+		"definition": jsonObject{"max-length": 1000},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	policies, err := service.listPolicies(jsonObject{"virtual_host": vhost})
+	if err != nil || !containsNamedItem(policies.(jsonObject)["policies"].([]jsonObject), policy) {
+		t.Fatalf("unexpected policies %#v, %v", policies, err)
+	}
+	if _, err := service.deletePolicy(jsonObject{"virtual_host": vhost, "name": policy}); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := service.createUser(jsonObject{"name": user, "password": "temporary-password", "tags": []any{"management"}}); err != nil {
+		t.Fatal(err)
+	}
+	users, err := service.listUsers(jsonObject{})
+	if err != nil || !containsNamedItem(users.(jsonObject)["users"].([]jsonObject), user) {
+		t.Fatalf("unexpected users %#v, %v", users, err)
+	}
+	if _, err := service.grantPermission(jsonObject{"user": user, "virtual_host": vhost}); err != nil {
+		t.Fatal(err)
+	}
+	permissions, err := service.listPermissions(jsonObject{"user": user, "virtual_host": vhost})
+	if err != nil || len(permissions.(jsonObject)["permissions"].([]jsonObject)) != 1 {
+		t.Fatalf("unexpected permissions %#v, %v", permissions, err)
+	}
+	if _, err := service.revokePermission(jsonObject{"user": user, "virtual_host": vhost}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.deleteUser(jsonObject{"name": user}); err != nil {
+		t.Fatal(err)
+	}
+
+	namespaces, err := service.listNamespaces(jsonObject{})
+	if err != nil || !containsNamedItem(namespaces.(jsonObject)["namespaces"].([]jsonObject), vhost) {
+		t.Fatalf("unexpected namespaces %#v, %v", namespaces, err)
+	}
+	connections, err := service.listClientConnections(jsonObject{"all_vhosts": true})
+	if err != nil || len(connections.(jsonObject)["connections"].([]jsonObject)) == 0 {
+		t.Fatalf("unexpected connections %#v, %v", connections, err)
+	}
+	channels, err := service.listClientChannels(jsonObject{"all_vhosts": true})
+	if err != nil || len(channels.(jsonObject)["channels"].([]jsonObject)) == 0 {
+		t.Fatalf("unexpected channels %#v, %v", channels, err)
+	}
+	if _, err := service.getOverview(jsonObject{}); err != nil {
+		t.Fatal(err)
+	}
+	nodes, err := service.listNodes(jsonObject{})
+	if err != nil || len(nodes.(jsonObject)["nodes"].([]jsonObject)) == 0 {
+		t.Fatalf("unexpected nodes %#v, %v", nodes, err)
+	}
+	cluster, err := service.describeCluster(jsonObject{})
+	if err != nil || cluster.(jsonObject)["version"] == nil {
+		t.Fatalf("unexpected cluster %#v, %v", cluster, err)
+	}
+
+	purged, err := service.purgeQueue(jsonObject{"topic": queue, "virtual_host": vhost})
+	if err != nil || purged.(jsonObject)["purged"] != 1 {
+		t.Fatalf("unexpected purge %#v, %v", purged, err)
+	}
+	if _, err := service.unbind(jsonObject{
+		"source": exchange, "destination": queue, "destinationType": "queue", "routingKey": "orders.*", "virtual_host": vhost,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.deleteTopic(jsonObject{"topic": queue, "virtual_host": vhost}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := service.deleteExchange(jsonObject{"name": exchange, "virtual_host": vhost}); err != nil {
+		t.Fatal(err)
+	}
+	if client := service.vhostClients[vhost]; client != nil {
+		client.close()
+		delete(service.vhostClients, vhost)
+	}
+	if _, err := service.deleteNamespace(jsonObject{"namespace": vhost}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func containsNamedItem(items []jsonObject, name string) bool {
+	for _, item := range items {
+		if stringOrEmpty(item, "name") == name {
+			return true
+		}
+	}
+	return false
+}
+
+func envOrDefault(key, fallback string) string {
+	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
+		return value
+	}
+	return fallback
+}
+
+func envIntOrDefault(t *testing.T, key string, fallback int) int {
+	t.Helper()
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil {
+		t.Fatalf("invalid %s: %v", key, err)
+	}
+	return parsed
+}

agents/drivers/rabbitmq/main.go

@@ -0,0 +1,529 @@
+package main
+
+import (
+	"bufio"
+	"crypto/tls"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const (
+	protocolVersion         = 1
+	agentProtocolVersion    = 1
+	defaultAMQPPort         = 5672
+	defaultRequestTimeout   = 30 * time.Second
+	defaultHandshakeTimeout = 10 * time.Second
+	defaultHeartbeat        = 60 * time.Second
+	defaultChannelMax       = 2047
+	maxRPCMessageBytes      = 32 * 1024 * 1024
+)
+
+var capabilities = []string{
+	"mq_connect", "mq_test_connection", "mq_topics",
+	"mq_messages", "mq_config", "mq_monitoring", "mq_exchanges",
+	"mq_client_connections", "mq_user_permissions", "mq_policies",
+}
+
+var allVhostsUnsupportedMethods = map[string]struct{}{
+	"mq_create_topic": {}, "mq_delete_topic": {}, "mq_purge_queue": {}, "mq_send_message": {},
+	"mq_bind": {}, "mq_unbind": {}, "mq_create_exchange": {}, "mq_delete_exchange": {},
+	"mq_peek_messages": {}, "mq_get_topic_stats": {}, "mq_list_consumers": {}, "mq_close_connection": {},
+	"mq_grant_permission": {}, "mq_revoke_permission": {}, "mq_set_policy": {}, "mq_delete_policy": {},
+}
+
+type jsonObject map[string]any
+
+type rpcRequest struct {
+	ID     json.RawMessage `json:"id"`
+	Method string          `json:"method"`
+	Params json.RawMessage `json:"params"`
+}
+
+type rpcError struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+type rpcResponse struct {
+	JSONRPC string          `json:"jsonrpc"`
+	ID      json.RawMessage `json:"id"`
+	Result  any             `json:"result,omitempty"`
+	Error   *rpcError       `json:"error,omitempty"`
+}
+
+type handshakeResult struct {
+	ProtocolVersion      int      `json:"protocolVersion"`
+	AgentProtocolVersion int      `json:"agentProtocolVersion"`
+	Capabilities         []string `json:"capabilities"`
+}
+
+type address struct {
+	Host string
+	Port int
+}
+
+type vhostClient struct {
+	connection *amqp.Connection
+	channel    *amqp.Channel
+}
+
+type server struct {
+	connection       *amqp.Connection
+	channel          *amqp.Channel
+	cachedConnection jsonObject
+	vhostClients     map[string]*vhostClient
+}
+
+func main() {
+	service := newServer()
+	encoder := json.NewEncoder(os.Stdout)
+	encoder.SetEscapeHTML(false)
+	fmt.Fprintln(os.Stdout, `{"ready":true}`)
+
+	scanner := bufio.NewScanner(os.Stdin)
+	scanner.Buffer(make([]byte, 64*1024), maxRPCMessageBytes)
+	for scanner.Scan() {
+		response, shutdown := service.handleRequest(scanner.Bytes())
+		if err := encoder.Encode(response); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			return
+		}
+		if shutdown {
+			return
+		}
+	}
+	service.closeClients()
+}
+
+func newServer() *server {
+	return &server{vhostClients: make(map[string]*vhostClient)}
+}
+
+func (s *server) handleRequest(line []byte) (rpcResponse, bool) {
+	response := rpcResponse{JSONRPC: "2.0", ID: json.RawMessage("null")}
+	var request rpcRequest
+	if err := json.Unmarshal(line, &request); err != nil {
+		response.Error = &rpcError{Code: -1, Message: normalizeErrorMessage(err)}
+		return response, false
+	}
+	if len(request.ID) > 0 {
+		response.ID = request.ID
+	}
+	params := jsonObject{}
+	if len(request.Params) > 0 && string(request.Params) != "null" {
+		var decoded any
+		if err := decodeJSON(request.Params, &decoded); err != nil {
+			response.Error = &rpcError{Code: -1, Message: normalizeErrorMessage(err)}
+			return response, false
+		}
+		if object, ok := decoded.(map[string]any); ok {
+			params = jsonObject(object)
+		}
+	}
+	result, shutdown, err := s.dispatch(request.Method, params)
+	if err != nil {
+		response.Error = &rpcError{Code: -1, Message: normalizeErrorMessage(err)}
+		return response, false
+	}
+	response.Result = result
+	return response, shutdown
+}
+
+func (s *server) dispatch(method string, params jsonObject) (any, bool, error) {
+	if _, unsupported := allVhostsUnsupportedMethods[method]; unsupported && allVhostsRequested(params) {
+		return nil, false, errors.New("all_vhosts is only supported for list operations")
+	}
+	switch method {
+	case "handshake":
+		return handshakeResult{protocolVersion, agentProtocolVersion, capabilities}, false, nil
+	case "connect":
+		result, err := s.connect(params)
+		return result, false, err
+	case "test_connection":
+		result, err := s.testConnection(params)
+		return result, false, err
+	case "disconnect":
+		s.closeClients()
+		return okResult(), false, nil
+	case "shutdown":
+		s.closeClients()
+		return okResult(), true, nil
+	case "mq_list_topics":
+		result, err := s.listTopics(params)
+		return result, false, err
+	case "mq_create_topic":
+		result, err := s.createTopic(params)
+		return result, false, err
+	case "mq_delete_topic":
+		result, err := s.deleteTopic(params)
+		return result, false, err
+	case "mq_get_topic_stats":
+		result, err := s.getTopicStats(params)
+		return result, false, err
+	case "mq_get_topic_config":
+		result, err := s.getTopicConfig(params)
+		return result, false, err
+	case "mq_alter_topic_config":
+		return nil, false, errors.New("RabbitMQ queue arguments are immutable after declaration; delete and re-declare the queue to change them")
+	case "mq_purge_queue":
+		result, err := s.purgeQueue(params)
+		return result, false, err
+	case "mq_list_consumers":
+		result, err := s.listConsumers(params)
+		return result, false, err
+	case "mq_list_namespaces":
+		result, err := s.listNamespaces(params)
+		return result, false, err
+	case "mq_create_namespace":
+		result, err := s.createNamespace(params)
+		return result, false, err
+	case "mq_delete_namespace":
+		result, err := s.deleteNamespace(params)
+		return result, false, err
+	case "mq_list_exchanges":
+		result, err := s.listExchanges(params)
+		return result, false, err
+	case "mq_create_exchange":
+		result, err := s.createExchange(params)
+		return result, false, err
+	case "mq_delete_exchange":
+		result, err := s.deleteExchange(params)
+		return result, false, err
+	case "mq_list_bindings":
+		result, err := s.listBindings(params)
+		return result, false, err
+	case "mq_bind":
+		result, err := s.bind(params)
+		return result, false, err
+	case "mq_unbind":
+		result, err := s.unbind(params)
+		return result, false, err
+	case "mq_list_connections":
+		result, err := s.listClientConnections(params)
+		return result, false, err
+	case "mq_list_channels":
+		result, err := s.listClientChannels(params)
+		return result, false, err
+	case "mq_close_connection":
+		result, err := s.closeClientConnection(params)
+		return result, false, err
+	case "mq_list_users":
+		result, err := s.listUsers(params)
+		return result, false, err
+	case "mq_create_user":
+		result, err := s.createUser(params)
+		return result, false, err
+	case "mq_delete_user":
+		result, err := s.deleteUser(params)
+		return result, false, err
+	case "mq_list_permissions":
+		result, err := s.listPermissions(params)
+		return result, false, err
+	case "mq_grant_permission":
+		result, err := s.grantPermission(params)
+		return result, false, err
+	case "mq_revoke_permission":
+		result, err := s.revokePermission(params)
+		return result, false, err
+	case "mq_list_policies":
+		result, err := s.listPolicies(params)
+		return result, false, err
+	case "mq_set_policy":
+		result, err := s.setPolicy(params)
+		return result, false, err
+	case "mq_delete_policy":
+		result, err := s.deletePolicy(params)
+		return result, false, err
+	case "mq_peek_messages":
+		result, err := s.peekMessages(params)
+		return result, false, err
+	case "mq_send_message":
+		result, err := s.sendMessage(params)
+		return result, false, err
+	case "mq_describe_cluster":
+		result, err := s.describeCluster(params)
+		return result, false, err
+	case "mq_overview":
+		result, err := s.getOverview(params)
+		return result, false, err
+	case "mq_list_nodes":
+		result, err := s.listNodes(params)
+		return result, false, err
+	default:
+		return nil, false, fmt.Errorf("Unknown method: %s", method)
+	}
+}
+
+func (s *server) connect(params jsonObject) (any, error) {
+	config := connectionObject(params)
+	nextConnection, err := openConnection(config)
+	if err != nil {
+		return nil, err
+	}
+	nextChannel, err := nextConnection.Channel()
+	if err != nil {
+		closeConnection(nextConnection)
+		return nil, err
+	}
+	s.closeClients()
+	s.connection = nextConnection
+	s.channel = nextChannel
+	s.cachedConnection = deepCopyObject(config)
+	return okResult(), nil
+}
+
+func (s *server) testConnection(params jsonObject) (any, error) {
+	config := connectionObject(params)
+	connection, err := openConnection(config)
+	if err != nil {
+		return nil, err
+	}
+	defer closeConnection(connection)
+	version := serverString(connection.Properties, "version")
+	return jsonObject{
+		"ok":            true,
+		"product":       serverString(connection.Properties, "product"),
+		"version":       version,
+		"serverVersion": version,
+		"clusterName":   serverString(connection.Properties, "cluster_name"),
+		"platform":      serverString(connection.Properties, "platform"),
+	}, nil
+}
+
+func (s *server) closeClients() {
+	for key, client := range s.vhostClients {
+		client.close()
+		delete(s.vhostClients, key)
+	}
+	closeChannel(s.channel)
+	s.channel = nil
+	closeConnection(s.connection)
+	s.connection = nil
+	s.cachedConnection = nil
+}
+
+func (client *vhostClient) close() {
+	if client == nil {
+		return
+	}
+	closeChannel(client.channel)
+	closeConnection(client.connection)
+}
+
+func (client *vhostClient) isOpen() bool {
+	return client != nil && client.connection != nil && !client.connection.IsClosed() && client.channel != nil && !client.channel.IsClosed()
+}
+
+func closeChannel(channel *amqp.Channel) {
+	if channel != nil {
+		_ = channel.Close()
+	}
+}
+
+func closeConnection(connection *amqp.Connection) {
+	if connection != nil {
+		_ = connection.Close()
+	}
+}
+
+func openConnection(config jsonObject) (*amqp.Connection, error) {
+	addresses, err := resolveAddresses(config)
+	if err != nil {
+		return nil, err
+	}
+	var lastError error
+	for _, endpoint := range addresses {
+		connection, dialError := dialAddress(config, endpoint)
+		if dialError == nil {
+			return connection, nil
+		}
+		lastError = dialError
+	}
+	if lastError == nil {
+		return nil, errors.New("addresses is required")
+	}
+	return nil, lastError
+}
+
+func dialAddress(config jsonObject, endpoint address) (*amqp.Connection, error) {
+	properties := objectOrNil(config, "properties")
+	connectionTimeout := durationMilliseconds(config, "request_timeout_ms", defaultRequestTimeout)
+	if configured, ok := integerProperty(properties, "connection_timeout_ms"); ok {
+		connectionTimeout = time.Duration(configured) * time.Millisecond
+	}
+	handshakeTimeout := defaultHandshakeTimeout
+	if configured, ok := integerProperty(properties, "handshake_timeout_ms"); ok {
+		handshakeTimeout = time.Duration(configured) * time.Millisecond
+	}
+	heartbeat := int(defaultHeartbeat / time.Second)
+	if configured, ok := integerProperty(properties, "requested_heartbeat"); ok {
+		heartbeat = configured
+	}
+	scheme := "amqp"
+	if amqpTLSEnabled(config) {
+		scheme = "amqps"
+	}
+	uri := url.URL{Scheme: scheme, Host: net.JoinHostPort(endpoint.Host, strconv.Itoa(endpoint.Port)), Path: "/"}
+	query := uri.Query()
+	query.Set("heartbeat", strconv.Itoa(heartbeat))
+	uri.RawQuery = query.Encode()
+	amqpConfig := amqp.Config{
+		SASL: []amqp.Authentication{&amqp.PlainAuth{
+			Username: credentialOrGuest(config, "username"),
+			Password: credentialOrGuest(config, "password"),
+		}},
+		Vhost:      stringOrDefault(config, "virtual_host", "/"),
+		ChannelMax: defaultChannelMax,
+		Heartbeat:  time.Duration(heartbeat) * time.Second,
+		Dial: func(network, target string) (net.Conn, error) {
+			dialer := net.Dialer{Timeout: connectionTimeout}
+			connection, err := dialer.Dial(network, target)
+			if err != nil {
+				return nil, err
+			}
+			if err := connection.SetDeadline(time.Now().Add(handshakeTimeout)); err != nil {
+				_ = connection.Close()
+				return nil, err
+			}
+			return connection, nil
+		},
+	}
+	if scheme == "amqps" {
+		amqpConfig.TLSClientConfig = &tls.Config{
+			ServerName:         endpoint.Host,
+			InsecureSkipVerify: tlsSkipVerify(config),
+		}
+	}
+	return amqp.DialConfig(uri.String(), amqpConfig)
+}
+
+func (s *server) channelFor(params jsonObject) (*amqp.Channel, error) {
+	defaultVhost := "/"
+	if s.cachedConnection != nil {
+		defaultVhost = stringOrDefault(s.cachedConnection, "virtual_host", "/")
+	}
+	vhost := effectiveVhost(params, s.cachedConnection)
+	if vhost == defaultVhost {
+		return s.primaryChannel()
+	}
+	if s.cachedConnection == nil {
+		return nil, errors.New("Not connected. Call connect first.")
+	}
+	if client := s.vhostClients[vhost]; client != nil && client.isOpen() {
+		return client.channel, nil
+	} else if client != nil {
+		client.close()
+		delete(s.vhostClients, vhost)
+	}
+	config := deepCopyObject(s.cachedConnection)
+	config["virtual_host"] = vhost
+	connection, err := openConnection(config)
+	if err != nil {
+		return nil, err
+	}
+	channel, err := connection.Channel()
+	if err != nil {
+		closeConnection(connection)
+		return nil, err
+	}
+	s.vhostClients[vhost] = &vhostClient{connection: connection, channel: channel}
+	return channel, nil
+}
+
+func (s *server) primaryChannel() (*amqp.Channel, error) {
+	if s.connection != nil && !s.connection.IsClosed() && !needsNewChannel(s.channel) {
+		return s.channel, nil
+	}
+	if s.connection == nil || s.connection.IsClosed() {
+		if s.cachedConnection == nil {
+			return nil, errors.New("Not connected. Call connect first.")
+		}
+		closeConnection(s.connection)
+		connection, err := openConnection(s.cachedConnection)
+		if err != nil {
+			return nil, err
+		}
+		s.connection = connection
+	}
+	closeChannel(s.channel)
+	channel, err := s.connection.Channel()
+	if err != nil {
+		return nil, err
+	}
+	s.channel = channel
+	return channel, nil
+}
+
+func needsNewChannel(channel *amqp.Channel) bool {
+	return channel == nil || channel.IsClosed()
+}
+
+func resolveAddresses(config jsonObject) ([]address, error) {
+	addresses := strings.TrimSpace(stringOrEmpty(config, "addresses"))
+	if addresses == "" {
+		addresses = strings.TrimSpace(stringOrEmpty(config, "host"))
+	}
+	if addresses == "" {
+		return nil, errors.New("addresses is required")
+	}
+	return parseAddresses(addresses, intOrDefault(config, "port", defaultAMQPPort))
+}
+
+func parseAddresses(value string, defaultPort int) ([]address, error) {
+	result := make([]address, 0)
+	for _, part := range strings.Split(value, ",") {
+		trimmed := strings.TrimSpace(part)
+		if trimmed == "" {
+			continue
+		}
+		host := trimmed
+		port := defaultPort
+		if parsedHost, parsedPort, err := net.SplitHostPort(trimmed); err == nil {
+			host = parsedHost
+			parsed, parseError := strconv.Atoi(parsedPort)
+			if parseError != nil {
+				return nil, parseError
+			}
+			port = parsed
+		} else if colon := strings.LastIndex(trimmed, ":"); colon > 0 && colon < len(trimmed)-1 && strings.Count(trimmed, ":") == 1 {
+			parsed, parseError := strconv.Atoi(trimmed[colon+1:])
+			if parseError != nil {
+				return nil, parseError
+			}
+			host = trimmed[:colon]
+			port = parsed
+		} else {
+			host = strings.TrimPrefix(strings.TrimSuffix(trimmed, "]"), "[")
+		}
+		result = append(result, address{Host: host, Port: port})
+	}
+	if len(result) == 0 {
+		return nil, errors.New("addresses is required")
+	}
+	return result, nil
+}
+
+func amqpTLSEnabled(config jsonObject) bool {
+	_, nestedTLS := config["tls"].(map[string]any)
+	if !nestedTLS {
+		_, nestedTLS = config["tls"].(jsonObject)
+	}
+	return nestedTLS || boolOrDefault(config, "tls_skip_verify", false) || boolProperty(config, "ssl") || boolProperty(config, "tls")
+}
+
+func tlsSkipVerify(config jsonObject) bool {
+	if boolOrDefault(config, "tls_skip_verify", false) {
+		return true
+	}
+	tlsConfig := objectOrNil(config, "tls")
+	return boolOrDefault(tlsConfig, "skip_verify", false)
+}

agents/drivers/rabbitmq/management.go

@@ -0,0 +1,248 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	defaultManagementPort    = 15672
+	defaultManagementTLSPort = 15671
+	managementPageSize       = 100
+	managementConnectTimeout = 10 * time.Second
+	managementRequestTimeout = 20 * time.Second
+)
+
+type managementStatusError struct {
+	status int
+	method string
+	path   string
+}
+
+func (err *managementStatusError) Error() string {
+	return managementErrorMessage(err.status, err.method, err.path)
+}
+
+func managementGet(connection jsonObject, path string) (any, error) {
+	return managementRequest(connection, http.MethodGet, path, nil)
+}
+
+func managementSend(connection jsonObject, method, path string, body jsonObject) (any, error) {
+	return managementRequest(connection, method, path, body)
+}
+
+func managementRequest(connection jsonObject, method, path string, body jsonObject) (any, error) {
+	baseURLs, err := managementBaseURLs(connection)
+	if err != nil {
+		return nil, err
+	}
+	var lastConnectionError error
+	for _, baseURL := range baseURLs {
+		result, requestError := managementRequestOnce(baseURL, connection, method, path, body)
+		if requestError == nil {
+			return result, nil
+		}
+		var statusError *managementStatusError
+		if errors.As(requestError, &statusError) {
+			return nil, requestError
+		}
+		var networkError net.Error
+		if errors.As(requestError, &networkError) {
+			lastConnectionError = requestError
+			continue
+		}
+		return nil, requestError
+	}
+	if lastConnectionError != nil {
+		return nil, lastConnectionError
+	}
+	return nil, errors.New("No management API endpoint candidates")
+}
+
+func managementRequestOnce(baseURL string, connection jsonObject, method, path string, body jsonObject) (any, error) {
+	var requestBody io.Reader
+	if body != nil {
+		encoded, err := json.Marshal(body)
+		if err != nil {
+			return nil, err
+		}
+		requestBody = bytes.NewReader(encoded)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), managementRequestTimeout)
+	defer cancel()
+	request, err := http.NewRequestWithContext(ctx, method, baseURL+path, requestBody)
+	if err != nil {
+		return nil, err
+	}
+	request.Header.Set("Authorization", basicAuthHeader(
+		credentialOrGuest(connection, "username"), credentialOrGuest(connection, "password")))
+	if body != nil {
+		request.Header.Set("Content-Type", "application/json")
+	}
+	transport := &http.Transport{
+		DialContext:           (&net.Dialer{Timeout: managementConnectTimeout}).DialContext,
+		TLSHandshakeTimeout:   managementConnectTimeout,
+		ResponseHeaderTimeout: managementConnectTimeout,
+		TLSClientConfig:       &tls.Config{InsecureSkipVerify: tlsSkipVerify(connection)},
+	}
+	defer transport.CloseIdleConnections()
+	client := &http.Client{Transport: transport}
+	response, err := client.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return nil, &managementStatusError{status: response.StatusCode, method: method, path: path}
+	}
+	if response.StatusCode == http.StatusNoContent {
+		return nil, nil
+	}
+	data, err := io.ReadAll(response.Body)
+	if err != nil {
+		return nil, err
+	}
+	if strings.TrimSpace(string(data)) == "" {
+		return nil, nil
+	}
+	var result any
+	if err := decodeJSON(data, &result); err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
+func managementGetAll(connection jsonObject, path string) ([]any, error) {
+	all := make([]any, 0)
+	for page := 1; ; page++ {
+		separator := "?"
+		if strings.Contains(path, "?") {
+			separator = "&"
+		}
+		response, err := managementGet(connection,
+			path+separator+"page="+strconv.Itoa(page)+"&page_size="+strconv.Itoa(managementPageSize))
+		if err != nil {
+			return nil, err
+		}
+		switch typed := response.(type) {
+		case []any:
+			return append(all, typed...), nil
+		case map[string]any:
+			items, exists := typed["items"]
+			if !exists {
+				return nil, fmt.Errorf("Unexpected management API response for list endpoint %s", path)
+			}
+			if array, ok := items.([]any); ok {
+				all = append(all, array...)
+			}
+			pageCount := integerOrNull(jsonObject(typed), "page_count")
+			if pageCount == nil || page >= *pageCount {
+				return all, nil
+			}
+		default:
+			return nil, fmt.Errorf("Unexpected management API response for list endpoint %s", path)
+		}
+	}
+}
+
+func managementBaseURLs(connection jsonObject) ([]string, error) {
+	if explicit := stringOrNull(connection, "management_url"); explicit != nil && strings.TrimSpace(*explicit) != "" {
+		return []string{normalizeManagementURL(*explicit)}, nil
+	}
+	tlsEnabled := managementTLS(connection)
+	port := managementPort(connection, tlsEnabled)
+	addresses, err := resolveAddresses(connection)
+	if err != nil {
+		return nil, err
+	}
+	baseURLs := make([]string, 0, len(addresses))
+	for _, endpoint := range addresses {
+		baseURLs = append(baseURLs, managementBaseURL(endpoint.Host, port, tlsEnabled))
+	}
+	return baseURLs, nil
+}
+
+func managementBaseURL(host string, port int, tlsEnabled bool) string {
+	scheme := "http"
+	if tlsEnabled {
+		scheme = "https"
+	}
+	return scheme + "://" + net.JoinHostPort(host, strconv.Itoa(port))
+}
+
+func normalizeManagementURL(value string) string {
+	return strings.TrimRight(strings.TrimSpace(value), "/")
+}
+
+func managementTLS(connection jsonObject) bool {
+	return objectOrNil(connection, "tls") != nil || boolProperty(connection, "ssl") || boolProperty(connection, "tls")
+}
+
+func managementPort(connection jsonObject, tlsEnabled bool) int {
+	if configured, ok := integerProperty(objectOrNil(connection, "properties"), "management_port"); ok {
+		return configured
+	}
+	if tlsEnabled {
+		return defaultManagementTLSPort
+	}
+	return defaultManagementPort
+}
+
+func credentialOrGuest(connection jsonObject, key string) string {
+	value := stringOrNull(connection, key)
+	if value == nil || strings.TrimSpace(*value) == "" {
+		return "guest"
+	}
+	return *value
+}
+
+func basicAuthHeader(username, password string) string {
+	return "Basic " + base64.StdEncoding.EncodeToString([]byte(username+":"+password))
+}
+
+func managementErrorMessage(status int, method, path string) string {
+	base := fmt.Sprintf("RabbitMQ management API returned HTTP %d for %s %s.", status, method, path)
+	if status == http.StatusUnauthorized || status == http.StatusForbidden {
+		return base + " Hint: check the username/password and that the user has a management permission tag (management, policymaker, monitoring, or administrator)."
+	}
+	return base + " The rabbitmq_management plugin must be enabled for this operation."
+}
+
+func urlEncodeVhost(value string) string {
+	return javaFormPathEscape(value)
+}
+
+func urlEncodePathSegment(value string) string {
+	return javaFormPathEscape(value)
+}
+
+func urlEncodeName(value string) string {
+	return urlEncodePathSegment(value)
+}
+
+func javaFormPathEscape(value string) string {
+	const hex = "0123456789ABCDEF"
+	var builder strings.Builder
+	for _, current := range []byte(value) {
+		if (current >= 'a' && current <= 'z') || (current >= 'A' && current <= 'Z') ||
+			(current >= '0' && current <= '9') || current == '-' || current == '_' || current == '.' || current == '*' {
+			builder.WriteByte(current)
+			continue
+		}
+		builder.WriteByte('%')
+		builder.WriteByte(hex[current>>4])
+		builder.WriteByte(hex[current&15])
+	}
+	return builder.String()
+}

agents/drivers/rabbitmq/management_test.go

@@ -0,0 +1,221 @@
+package main
+
+import (
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+)
+
+func TestManagementBaseURLs(t *testing.T) {
+	explicit, err := managementBaseURLs(mustObject(t, `{"management_url":" https://proxy:8443/rmq/ "}`))
+	if err != nil || len(explicit) != 1 || explicit[0] != "https://proxy:8443/rmq" {
+		t.Fatalf("unexpected explicit URLs %#v, %v", explicit, err)
+	}
+	withoutAddresses, err := managementBaseURLs(mustObject(t, `{"management_url":"http://mgmt:15672"}`))
+	if err != nil || withoutAddresses[0] != "http://mgmt:15672" {
+		t.Fatalf("unexpected URL %#v, %v", withoutAddresses, err)
+	}
+	derived, err := managementBaseURLs(mustObject(t, `{"addresses":"mq1:5672,mq2:5673"}`))
+	if err != nil || len(derived) != 2 || derived[0] != "http://mq1:15672" || derived[1] != "http://mq2:15672" {
+		t.Fatalf("unexpected derived URLs %#v, %v", derived, err)
+	}
+	tlsDerived, err := managementBaseURLs(mustObject(t, `{"addresses":"mq1","tls":{}}`))
+	if err != nil || tlsDerived[0] != "https://mq1:15671" {
+		t.Fatalf("unexpected TLS URLs %#v, %v", tlsDerived, err)
+	}
+	skipVerify, err := managementBaseURLs(mustObject(t, `{"addresses":"mq1","tls_skip_verify":true}`))
+	if err != nil || skipVerify[0] != "http://mq1:15672" {
+		t.Fatalf("unexpected skip-verify URLs %#v, %v", skipVerify, err)
+	}
+}
+
+func TestManagementErrorMessages(t *testing.T) {
+	for _, status := range []int{401, 403} {
+		message := managementErrorMessage(status, http.MethodGet, "/api/queues")
+		if !strings.Contains(message, "management permission tag") || strings.Contains(message, "plugin must be enabled") {
+			t.Fatalf("unexpected message %q", message)
+		}
+	}
+	message := managementErrorMessage(404, http.MethodGet, "/api/queues/%2F/gone")
+	if !strings.Contains(message, "plugin must be enabled") || strings.Contains(message, "management permission tag") {
+		t.Fatalf("unexpected message %q", message)
+	}
+}
+
+func TestManagementRequestSurfacesCredentialError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.WriteHeader(http.StatusUnauthorized)
+	}))
+	defer server.Close()
+	connection := jsonObject{"management_url": server.URL}
+	_, err := managementGet(connection, "/api/queues")
+	if err == nil || !strings.Contains(err.Error(), "HTTP 401") || !strings.Contains(err.Error(), "management permission tag") {
+		t.Fatalf("unexpected error %v", err)
+	}
+}
+
+func TestManagementGetAllPagination(t *testing.T) {
+	var mutex sync.Mutex
+	requestedPages := make([]int, 0)
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		page, _ := strconv.Atoi(request.URL.Query().Get("page"))
+		mutex.Lock()
+		requestedPages = append(requestedPages, page)
+		mutex.Unlock()
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(writer, `{"items":[{"name":"q`+strconv.Itoa(page)+`"}],"page":`+strconv.Itoa(page)+`,"page_count":3,"total_count":3}`)
+	}))
+	defer server.Close()
+	items, err := managementGetAll(jsonObject{"management_url": server.URL}, "/api/queues")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(items) != 3 || len(requestedPages) != 3 || requestedPages[0] != 1 || requestedPages[2] != 3 {
+		t.Fatalf("unexpected items %#v pages %#v", items, requestedPages)
+	}
+}
+
+func TestManagementGetAllPlainArray(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(writer, `[{"name":"guest"}]`)
+	}))
+	defer server.Close()
+	items, err := managementGetAll(jsonObject{"management_url": server.URL}, "/api/users")
+	if err != nil || len(items) != 1 {
+		t.Fatalf("unexpected items %#v, %v", items, err)
+	}
+}
+
+func TestManagementRequestFailsOverConnectionErrors(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.2:0")
+	if err != nil {
+		t.Skipf("secondary loopback address unavailable: %v", err)
+	}
+	port := listener.Addr().(*net.TCPAddr).Port
+	server := &http.Server{Handler: http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(writer, `[]`)
+	})}
+	defer server.Close()
+	go server.Serve(listener)
+	connection := mustObject(t, `{"addresses":"127.0.0.1,127.0.0.2","properties":{"management_port":`+strconv.Itoa(port)+`}}`)
+	response, err := managementGet(connection, "/api/queues")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := response.([]any); !ok {
+		t.Fatalf("unexpected response %#v", response)
+	}
+}
+
+func TestManagementHTTPErrorDoesNotFailOver(t *testing.T) {
+	first, second, port := pairedLoopbackServers(t,
+		http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			writer.WriteHeader(http.StatusNotFound)
+		}),
+		http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			writer.Header().Set("Content-Type", "application/json")
+			_, _ = io.WriteString(writer, `[]`)
+		}),
+	)
+	defer first.Close()
+	defer second.Close()
+	connection := mustObject(t, `{"addresses":"127.0.0.1,127.0.0.2","properties":{"management_port":`+strconv.Itoa(port)+`}}`)
+	_, err := managementGet(connection, "/api/queues")
+	if err == nil || !strings.Contains(err.Error(), "HTTP 404") {
+		t.Fatalf("unexpected error %v", err)
+	}
+}
+
+func TestManagementURLPathPrefix(t *testing.T) {
+	requestedPath := ""
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestedPath = request.URL.Path
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(writer, `[]`)
+	}))
+	defer server.Close()
+	_, err := managementGet(jsonObject{"management_url": server.URL + "/rmq/"}, "/api/queues")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if requestedPath != "/rmq/api/queues" {
+		t.Fatalf("got path %q", requestedPath)
+	}
+}
+
+func TestPolicyManagementOperations(t *testing.T) {
+	type capturedRequest struct {
+		Method string
+		Path   string
+		Body   jsonObject
+	}
+	requests := make([]capturedRequest, 0)
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		captured := capturedRequest{Method: request.Method, Path: request.URL.Path}
+		if request.Body != nil {
+			data, _ := io.ReadAll(request.Body)
+			if len(data) > 0 {
+				_ = decodeJSON(data, &captured.Body)
+			}
+		}
+		requests = append(requests, captured)
+		writer.Header().Set("Content-Type", "application/json")
+		switch request.Method {
+		case http.MethodGet:
+			_, _ = io.WriteString(writer, `[{"name":"ha","vhost":"/","pattern":"^ha","apply-to":"queues","priority":0,"definition":{"ha-mode":"all"}}]`)
+		default:
+			writer.WriteHeader(http.StatusNoContent)
+		}
+	}))
+	defer server.Close()
+	service := newServer()
+	service.cachedConnection = jsonObject{"management_url": server.URL, "username": "guest", "password": "guest"}
+	listed, err := service.listPolicies(jsonObject{"virtual_host": "/"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(listed.(jsonObject)["policies"].([]jsonObject)) != 1 {
+		t.Fatalf("unexpected policies %#v", listed)
+	}
+	_, err = service.setPolicy(mustObject(t, `{"virtual_host":"/","name":"ha","pattern":"^ha","definition":{"ha-mode":"all"}}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = service.deletePolicy(mustObject(t, `{"virtual_host":"/","name":"ha"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(requests) != 3 || requests[1].Method != http.MethodPut || requests[2].Method != http.MethodDelete {
+		t.Fatalf("unexpected requests %#v", requests)
+	}
+	if requests[1].Body["apply-to"] != "queues" || requests[1].Body["priority"] != json.Number("0") {
+		t.Fatalf("unexpected body %#v", requests[1].Body)
+	}
+}
+
+func pairedLoopbackServers(t *testing.T, firstHandler, secondHandler http.Handler) (*http.Server, *http.Server, int) {
+	t.Helper()
+	firstListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	port := firstListener.Addr().(*net.TCPAddr).Port
+	secondListener, err := net.Listen("tcp", "127.0.0.2:"+strconv.Itoa(port))
+	if err != nil {
+		firstListener.Close()
+		t.Skipf("secondary loopback address unavailable: %v", err)
+	}
+	first := &http.Server{Handler: firstHandler}
+	second := &http.Server{Handler: secondHandler}
+	go first.Serve(firstListener)
+	go second.Serve(secondListener)
+	return first, second, port
+}

agents/drivers/rabbitmq/mapping_test.go

@@ -0,0 +1,126 @@
+package main
+
+import "testing"
+
+func TestConsumersFromQueueInfo(t *testing.T) {
+	info := mustObject(t, `{
+      "consumer_details": [
+        {"consumer_tag":"ctag","active":true,"ack_required":true,"prefetch_count":25,"channel_details":{"name":"conn (1)"}},
+        "ignored"
+      ]
+    }`)
+	consumers := consumersFromQueueInfo(info)
+	if len(consumers) != 1 || consumers[0]["name"] != "conn (1)" || consumers[0]["tag"] != "ctag" || consumers[0]["prefetch"] != 25 {
+		t.Fatalf("unexpected consumers %#v", consumers)
+	}
+	if got := consumersFromQueueInfo(jsonObject{}); len(got) != 0 {
+		t.Fatalf("unexpected consumers %#v", got)
+	}
+}
+
+func TestExchangeAndBindingMappings(t *testing.T) {
+	defaultExchange := exchangeInfoFromJSON(mustObject(t, `{"name":"","type":"","durable":true,"auto_delete":false,"internal":false}`))
+	if defaultExchange["type"] != "default" || defaultExchange["durable"] != true {
+		t.Fatalf("unexpected exchange %#v", defaultExchange)
+	}
+	topicExchange := exchangeInfoFromJSON(mustObject(t, `{"name":"events","type":"topic","durable":true,"auto_delete":true,"internal":true}`))
+	if topicExchange["type"] != "topic" || topicExchange["autoDelete"] != true || topicExchange["internal"] != true {
+		t.Fatalf("unexpected exchange %#v", topicExchange)
+	}
+	binding := bindingInfoFromJSON(mustObject(t, `{
+      "source":"events","destination":"orders","destination_type":"queue","routing_key":"orders.*",
+      "arguments":{"x-priority":5,"alternate":true,"ignored":null}
+    }`))
+	if binding["destinationType"] != "queue" || binding["routingKey"] != "orders.*" {
+		t.Fatalf("unexpected binding %#v", binding)
+	}
+	arguments := binding["arguments"].(jsonObject)
+	if arguments["x-priority"] != int64(5) || arguments["alternate"] != true {
+		t.Fatalf("unexpected arguments %#v", arguments)
+	}
+	withoutArguments := bindingInfoFromJSON(mustObject(t, `{"source":"e","destination":"q","destination_type":"queue","routing_key":"","arguments":{}}`))
+	if _, exists := withoutArguments["arguments"]; exists {
+		t.Fatalf("unexpected arguments %#v", withoutArguments)
+	}
+}
+
+func TestConnectionAndChannelMappings(t *testing.T) {
+	connection := clientConnectionInfoFromJSON(mustObject(t, `{
+      "name":"127.0.0.1:1 -> 127.0.0.1:5672","user":"dbx","peer_host":"127.0.0.1","peer_port":1234,
+      "state":"running","channels":2,"recv_oct_details":{"rate":12.5},"send_oct_details":{"rate":8.25},"connected_at":1700000000000
+    }`))
+	if connection["recvRate"] != 12.5 || connection["sendRate"] != 8.25 || connection["connectedAt"] != int64(1700000000000) {
+		t.Fatalf("unexpected connection %#v", connection)
+	}
+	minimal := clientConnectionInfoFromJSON(mustObject(t, `{"name":"conn"}`))
+	for _, key := range []string{"recvRate", "sendRate", "connectedAt"} {
+		if _, exists := minimal[key]; exists {
+			t.Fatalf("unexpected %s in %#v", key, minimal)
+		}
+	}
+	channel := channelInfoFromJSON(mustObject(t, `{
+      "name":"conn (1)","connection_details":{"name":"conn"},"state":"running",
+      "prefetch_count":10,"messages_unacknowledged":4,"consumer_count":2
+    }`))
+	if channel["connectionName"] != "conn" || channel["messagesUnacked"] != int64(4) || channel["consumerCount"] != int64(2) {
+		t.Fatalf("unexpected channel %#v", channel)
+	}
+	if !channelMatchesConnection(channel, "conn") || !channelMatchesConnection(jsonObject{"name": "other (1)"}, "other") {
+		t.Fatal("connection matching failed")
+	}
+	if channelMatchesConnection(channel, "missing") {
+		t.Fatal("unexpected connection match")
+	}
+}
+
+func TestUserPermissionPolicyMappings(t *testing.T) {
+	user := userInfoFromJSON(mustObject(t, `{"name":"admin","tags":"administrator, management"}`))
+	if user["name"] != "admin" || len(user["tags"].([]string)) != 2 {
+		t.Fatalf("unexpected user %#v", user)
+	}
+	permission := permissionInfoFromJSON(mustObject(t, `{"user":"dbx","vhost":"/","configure":".*","write":"^orders","read":".*"}`))
+	if permission["write"] != "^orders" || permission["vhost"] != "/" {
+		t.Fatalf("unexpected permission %#v", permission)
+	}
+	policy := policyInfoFromJSON(mustObject(t, `{
+      "name":"ha","vhost":"/","pattern":"^ha","apply-to":"queues","priority":5,
+      "definition":{"ha-mode":"all","ha-sync-mode":"automatic","expires":60000,"ignored":null}
+    }`))
+	if policy["applyTo"] != "queues" || policy["priority"] != int64(5) {
+		t.Fatalf("unexpected policy %#v", policy)
+	}
+	definition := policy["definition"].(jsonObject)
+	if definition["expires"] != int64(60000) || definition["ha-mode"] != "all" {
+		t.Fatalf("unexpected definition %#v", definition)
+	}
+}
+
+func TestOverviewAndNodeMappings(t *testing.T) {
+	overview := overviewInfoFromJSON(mustObject(t, `{
+      "queue_totals":{"messages_ready":10,"messages_unacknowledged":2},
+      "message_stats":{"publish_details":{"rate":1.5},"deliver_get_details":{"rate":2.5},"ack_details":{"rate":3.5}},
+      "object_totals":{"queues":4,"exchanges":5,"connections":6,"channels":7,"consumers":8}
+    }`))
+	if overview["messagesReady"] != int64(10) || overview["publishRate"] != 1.5 || overview["totalConsumers"] != int64(8) {
+		t.Fatalf("unexpected overview %#v", overview)
+	}
+	minimal := overviewInfoFromJSON(jsonObject{})
+	if len(minimal) != 0 {
+		t.Fatalf("unexpected overview %#v", minimal)
+	}
+	node := nodeInfoFromJSON(mustObject(t, `{
+      "name":"rabbit@node","running":true,"mem_used":100,"mem_limit":200,"disk_free":300,
+      "fd_used":4,"fd_total":5,"sockets_used":6,"sockets_total":7,"uptime":8000
+    }`))
+	if node["running"] != true || node["memUsed"] != int64(100) || node["uptimeMs"] != int64(8000) {
+		t.Fatalf("unexpected node %#v", node)
+	}
+}
+
+func TestAttachVhost(t *testing.T) {
+	info := jsonObject{"name": "q"}
+	attachVhost(info, mustObject(t, `{"vhost":"orders"}`))
+	if info["vhost"] != "orders" {
+		t.Fatalf("unexpected info %#v", info)
+	}
+}

agents/scripts/driver_release_packages_test.py

@@ -20,6 +20,8 @@ class DriverReleasePackagesTest(unittest.TestCase):
             native_source.write_bytes(b"MZtest-agent")
             duckdb_source = release_dir / "dbx-agent-duckdb-macos-aarch64"
             duckdb_source.write_bytes(b"\xcf\xfa\xed\xfetest-duckdb-agent")
+            rabbitmq_source = release_dir / "dbx-agent-rabbitmq-linux-x64"
+            rabbitmq_source.write_bytes(b"\x7fELFtest-rabbitmq-agent")
             java_source = release_dir / "dbx-agent-h2.jar"
             java_source.write_bytes(b"test-jar")
             versions = {
@@ -28,13 +30,15 @@ class DriverReleasePackagesTest(unittest.TestCase):
                 "xugu": "0.1.20",
                 "kingbase": "0.1.34",
                 "duckdb": "0.1.0",
+                "rabbitmq": "0.1.0",
             }
 
             renamed = version_agent_artifacts(release_dir, versions)
             versioned_java = release_dir / "dbx-agent-h2-0.2.5.jar"
             versioned_native = release_dir / "dbx-agent-kingbase-0.1.34-windows-x64.exe"
             versioned_duckdb = release_dir / "dbx-agent-duckdb-0.1.0-macos-aarch64"
-            self.assertEqual(renamed, [versioned_java, versioned_native, versioned_duckdb])
+            versioned_rabbitmq = release_dir / "dbx-agent-rabbitmq-0.1.0-linux-x64"
+            self.assertEqual(renamed, [versioned_java, versioned_native, versioned_duckdb, versioned_rabbitmq])
 
             registry = {
                 "jres": {"21": {"version": "21", "platforms": {}}},
@@ -72,6 +76,19 @@ class DriverReleasePackagesTest(unittest.TestCase):
                             }
                         },
                     },
+                    "rabbitmq": {
+                        "version": "0.1.0",
+                        "label": "RabbitMQ",
+                        "min_app_version": "0.6.0",
+                        "jre": "21",
+                        "jar": {"url": "https://example.com/legacy-placeholder.jar", "size": 0},
+                        "native": {
+                            "linux-x64": {
+                                "url": f"https://example.com/{versioned_rabbitmq.name}",
+                                "size": versioned_rabbitmq.stat().st_size,
+                            }
+                        },
+                    },
                 },
             }
             (release_dir / "agent-registry.json").write_text(json.dumps(registry), encoding="utf-8")
@@ -84,12 +101,14 @@ class DriverReleasePackagesTest(unittest.TestCase):
                     release_dir / "dbx-agent-h2-0.2.5.tar.zst",
                     release_dir / "dbx-agent-kingbase-0.1.34-windows-x64.tar.zst",
                     release_dir / "dbx-agent-duckdb-0.1.0-macos-aarch64.tar.zst",
+                    release_dir / "dbx-agent-rabbitmq-0.1.0-linux-x64.tar.zst",
                 ],
             )
             package_cases = [
                 (outputs[0], "h2", versioned_java, "jar", None),
                 (outputs[1], "kingbase", versioned_native, "native", "windows-x64"),
                 (outputs[2], "duckdb", versioned_duckdb, "native", "macos-aarch64"),
+                (outputs[3], "rabbitmq", versioned_rabbitmq, "native", "linux-x64"),
             ]
             for output, driver_name, source, artifact_type, platform in package_cases:
                 tar_bytes = subprocess.run(
@@ -116,6 +135,7 @@ class DriverReleasePackagesTest(unittest.TestCase):
                 (final_registry["drivers"]["h2"]["jar"], outputs[0]),
                 (final_registry["drivers"]["kingbase"]["native"]["windows-x64"], outputs[1]),
                 (final_registry["drivers"]["duckdb"]["native"]["macos-aarch64"], outputs[2]),
+                (final_registry["drivers"]["rabbitmq"]["native"]["linux-x64"], outputs[3]),
             ]
             for artifact, output in release_artifacts:
                 self.assertEqual(artifact["url"], f"https://example.com/{output.name}")
@@ -124,7 +144,7 @@ class DriverReleasePackagesTest(unittest.TestCase):
                 self.assertEqual(len(artifact["sha256"]), 64)
 
             removed = remove_raw_driver_artifacts(release_dir)
-            self.assertEqual(removed, [versioned_duckdb, versioned_java, versioned_native])
+            self.assertEqual(removed, [versioned_duckdb, versioned_java, versioned_native, versioned_rabbitmq])
             self.assertTrue(all(output.is_file() for output in outputs))
 
     def test_full_offline_bundle_includes_supported_windows_artifacts(self) -> None:

agents/scripts/validate_agents.py

@@ -17,6 +17,7 @@ NATIVE_ONLY_AGENT_MODULES = {
     "oracle": "drivers/oracle-go",
     "kingbase": "drivers/kingbase-go",
     "xugu": "drivers/xugu",
+    "rabbitmq": "drivers/rabbitmq",
 }
 AUTO_VERSIONED_NATIVE_MODULES = {"duckdb"}
 JDBC_ARCHITECTURE_ALLOWLIST = {

agents/scripts/validate_agents_test.py

@@ -136,10 +136,10 @@ class ValidateAgentsTest(unittest.TestCase):
                 "include(*(infrastructureModules + driverModules))\n",
                 encoding="utf-8",
             )
-            for driver in ("oracle-go", "kingbase-go", "xugu", "duckdb"):
+            for driver in ("oracle-go", "kingbase-go", "xugu", "duckdb", "rabbitmq"):
                 (root / "drivers" / driver).mkdir(parents=True)
             (root / "versions.json").write_text(
-                json.dumps({"h2": "0.1.0", "oracle": "0.1.0", "kingbase": "0.1.0", "xugu": "0.1.0"}),
+                json.dumps({"h2": "0.1.0", "oracle": "0.1.0", "kingbase": "0.1.0", "xugu": "0.1.0", "rabbitmq": "0.1.0"}),
                 encoding="utf-8",
             )
 

agents/scripts/version_agent_artifacts.py

@@ -4,7 +4,7 @@ import json
 from pathlib import Path
 
 
-NATIVE_DRIVERS = ("oracle", "xugu", "kingbase", "duckdb")
+NATIVE_DRIVERS = ("oracle", "xugu", "kingbase", "duckdb", "rabbitmq")
 PLATFORMS = (
     "macos-aarch64",
     "macos-x64",

agents/settings.gradle

@@ -6,7 +6,7 @@ def driverModules = [
     'teradata', 'vertica', 'firebird', 'exasol', 'oceanbase-oracle', 'gbase8a', 'gbase8s',
     'bigquery', 'kylin', 'sundb', 'h2', 'h2-legacy', 'snowflake', 'trino', 'hive', 'spark',
     'db2', 'informix', 'neo4j', 'cassandra', 'mongodb', 'highgo', 'uxdb', 'tdengine', 'yashandb', 'oscar',
-    'iris', 'iotdb', 'etcd', 'zookeeper', 'kafka', 'rocketmq', 'rabbitmq', 'sqlserver-legacy'
+    'iris', 'iotdb', 'etcd', 'zookeeper', 'kafka', 'rocketmq', 'sqlserver-legacy'
 ]
 
 include(*(infrastructureModules + driverModules))

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

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

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

@@ -1,9 +1,9 @@
-//! RabbitMQ admin adapter. Communicates with a Java agent process
-//! (`RabbitMqAgent.java`) via JSON-RPC over stdin/stdout. The Java agent uses
-//! the `amqp-client` library for admin and message operations.
+//! RabbitMQ admin adapter. Communicates with the native Go agent via JSON-RPC
+//! over stdin/stdout. The agent uses `amqp091-go` for AMQP operations and the
+//! RabbitMQ management HTTP API for administrative operations.
 //!
 //! This adapter follows the same pattern as the Kafka agent:
-//! 1. Spawn a Java agent process via `AgentDriverClient`
+//! 1. Spawn the native agent process via `AgentDriverClient`
 //! 2. Perform JSON-RPC handshake + connect
 //! 3. Delegate all `MessageQueueAdmin` trait methods to JSON-RPC calls
 
@@ -59,7 +59,7 @@ pub struct RabbitMqAdmin {
 }
 
 impl RabbitMqAdmin {
-    /// Spawn the RabbitMQ Java agent, perform handshake, and connect.
+    /// Spawn the RabbitMQ native agent, perform handshake, and connect.
     pub async fn new(cfg: MqAdminConfig, launch: AgentLaunchSpec) -> Result<Self, String> {
         let mut client = AgentDriverClient::spawn(launch).await?;
 

docs/mq-quick-start.md

@@ -444,8 +444,9 @@ Agent 构建与安装：
 
 ```bash
 cd agents
-./gradlew :rabbitmq:shadowJar
-# 将 shadow JAR 安装到 DBX 数据目录 agents/drivers/rabbitmq/agent.jar
+cd drivers/rabbitmq
+go build -o agent .
+# 将原生 agent 安装到 DBX 数据目录 agents/drivers/rabbitmq/agent
 ```
 
 Docker 快速启动（AMQP 5672 + Management 15672，仅用于本地验证）：
@@ -506,4 +507,3 @@ const response = await mqRawRequest(connectionId, {
 })
 console.log(response.body)
 ```
-