feat(cli,broker): fleet agent list — fleet-wide names + presence diagnostic (#1553)

CHANGELOG.md

@@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased - Patch]
+## [Unreleased - Minor]
+
+### Added
+
+- `agent-relay fleet agent list [--pretty|--json] [--node <n>] [--all]` lists fleet-wide agents with their live, inventory, and roster presence, and reports unavailable node data explicitly.
 
 ### Fixed
 

crates/broker/src/listen_api.rs

@@ -80,6 +80,14 @@ pub enum ListenApiRequest {
     List {
         reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
     },
+    /// `GET /api/fleet-inventory` — snapshot of the in-process `fleet_inventory`
+    /// map (what the broker last published to the engine via `inventory.sync`).
+    /// Callers use this alongside `List` to detect the workers-vs-inventory
+    /// divergence documented in #1539 — an agent live in the PTY map that was
+    /// never (or is no longer) present in what the engine sees.
+    FleetInventory {
+        reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
+    },
     Threads {
         reply: tokio::sync::oneshot::Sender<Result<Value, String>>,
     },
@@ -445,6 +453,10 @@ fn listen_api_router_with_auth(
         .route("/api/session/renew", routing::post(listen_api_renew_lease))
         .route("/api/spawn", routing::post(listen_api_spawn))
         .route("/api/spawned", routing::get(listen_api_list))
+        .route(
+            "/api/fleet-inventory",
+            routing::get(listen_api_fleet_inventory),
+        )
         .route(
             "/api/spawned/{name}/model",
             routing::post(listen_api_set_model),
@@ -1172,6 +1184,24 @@ async fn listen_api_list(
     }
 }
 
+async fn listen_api_fleet_inventory(
+    axum::extract::State(state): axum::extract::State<ListenApiState>,
+) -> axum::Json<Value> {
+    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+    if state
+        .tx
+        .send(ListenApiRequest::FleetInventory { reply: reply_tx })
+        .await
+        .is_err()
+    {
+        return axum::Json(json!({ "success": false, "agents": [] }));
+    }
+    match reply_rx.await {
+        Ok(Ok(val)) => axum::Json(val),
+        _ => axum::Json(json!({ "success": false, "agents": [] })),
+    }
+}
+
 #[derive(Debug, Deserialize)]
 struct ListenApiSetModelPayload {
     model: String,
@@ -3893,6 +3923,76 @@ mod auth_tests {
         list_replier.await.expect("list replier should complete");
     }
 
+    #[tokio::test]
+    async fn fleet_inventory_route_forwards_and_returns_agents() {
+        // Must-fire: when the runtime reply carries an agents array, the HTTP
+        // response mirrors it verbatim. This is the diagnostic surface for
+        // #1553 / #1539 — an agent present in this map but absent from
+        // `/api/spawned` is exactly the divergence the CLI must flag.
+        let (router, mut rx) = test_router(Some("secret"));
+        let replier = tokio::spawn(async move {
+            if let Some(ListenApiRequest::FleetInventory { reply }) = rx.recv().await {
+                let _ = reply.send(Ok(json!({
+                    "node_name": "test-node",
+                    "agents": [
+                        {
+                            "agent_id": "ag_1",
+                            "name": "worker-a",
+                            "invocation_id": "inv_1"
+                        }
+                    ]
+                })));
+            }
+        });
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/fleet-inventory")
+                    .method("GET")
+                    .header("x-api-key", "secret")
+                    .body(Body::empty())
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = response_json(response).await;
+        assert_eq!(body["node_name"], "test-node");
+        assert_eq!(body["agents"][0]["name"], "worker-a");
+        assert_eq!(body["agents"][0]["agent_id"], "ag_1");
+
+        replier.await.expect("replier should complete");
+    }
+
+    #[tokio::test]
+    async fn fleet_inventory_route_returns_empty_agents_on_channel_close() {
+        // Must-not-fire: an unreachable runtime cannot invent phantom agents.
+        // Empty must not be conflated with "unknown" — the CLI relies on
+        // `success:false` + empty agents to distinguish this from a genuinely
+        // empty inventory.
+        let (router, rx) = test_router(Some("secret"));
+        drop(rx);
+
+        let response = router
+            .oneshot(
+                Request::builder()
+                    .uri("/api/fleet-inventory")
+                    .method("GET")
+                    .header("x-api-key", "secret")
+                    .body(Body::empty())
+                    .expect("request should build"),
+            )
+            .await
+            .expect("request should succeed");
+
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = response_json(response).await;
+        assert_eq!(body["success"], false);
+        assert_eq!(body["agents"], json!([]));
+    }
+
     #[tokio::test]
     async fn spawn_route_forwards_extended_fields() {
         let (router, mut rx) = test_router(Some("secret"));

crates/broker/src/runtime/api.rs

@@ -1354,6 +1354,17 @@ impl BrokerRuntime {
                     super::delivery::pending_message_counts(delivery_states, pending_deliveries);
                 let _ = reply.send(Ok(json!({ "agents": workers.list(&counts) })));
             }
+            ListenApiRequest::FleetInventory { reply } => {
+                // Report the in-process `fleet_inventory` map: the same
+                // snapshot the broker publishes to the engine via
+                // `inventory.sync`. Callers join this against `List` to
+                // detect the workers-vs-inventory divergence (#1539).
+                let agents: Vec<&InventoryAgent> = fleet_inventory.values().collect();
+                let _ = reply.send(Ok(json!({
+                    "node_name": fleet_node_name,
+                    "agents": agents,
+                })));
+            }
             ListenApiRequest::Threads { reply } => {
                 let mut messages: Vec<Value> = recent_thread_messages.iter().cloned().collect();
                 match relaycast_http.get_all_dms(200).await {

packages/cli/src/cli/bootstrap.test.ts

@@ -40,6 +40,7 @@ const expectedLeafCommands = [
   'reflex off',
   'reflex status',
   // fleet (serve is a hidden error stub, filtered out below)
+  'fleet agent list',
   'fleet config',
   'fleet disable',
   'fleet enable',