claw-store/src/cluster/rpc.rs

@@ -183,6 +183,16 @@ pub enum Method {
     /// Reply: 8 bytes (`u64` LE) on hit, single-byte
     /// [`ErrorCode::NotFound`] when no sidecar is present.
     GetTagExpiry = 0x1b,
+    /// Dashboard-v2 (2026-07-14): return a JSON snapshot of this
+    /// node's counts (blobs / tags / refs / snapshots / ref-tracking
+    /// + on-disk bytes). Feeds the fleet aggregator dashboard so an
+    /// operator gets a single-pane view of the fleet from any client
+    /// that can reach cluster RPC — no HTTP server on the daemon
+    /// nodes required.
+    ///
+    /// `payload`: empty.
+    /// Reply: JSON `DashboardStatusReply`.
+    DashboardStatus = 0x1c,
 }
 
 impl Method {
@@ -217,6 +227,7 @@ impl Method {
             0x19 => Some(Method::GetTagVersioned),
             0x1a => Some(Method::SetTagExpiry),
             0x1b => Some(Method::GetTagExpiry),
+            0x1c => Some(Method::DashboardStatus),
             _ => None,
         }
     }
@@ -297,6 +308,28 @@ pub struct PeerStatusReply {
     pub peers: Vec<PeerView>,
 }
 
+/// Dashboard-v2 aggregator payload. One per node. Wire: JSON.
+///
+/// Read by `serve_v2::AggregatorState` from any client that can
+/// reach cluster RPC (typically the operator's laptop running
+/// `claw-store serve --aggregator`).
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct DashboardStatusReply {
+    pub node_name: String,
+    pub zone: String,
+    pub blob_store_root: Option<String>,
+    pub blob_count: usize,
+    pub tag_count: usize,
+    pub ref_count: usize,
+    pub snapshot_count: usize,
+    pub ref_tracking_count: usize,
+    pub blob_store_bytes: u64,
+    /// `local_rustc_release` from PeerStatus, echoed here so the
+    /// aggregator doesn't need a second round-trip.
+    #[serde(default)]
+    pub rustc_release: Option<String>,
+}
+
 /// Reply payload for [`Method::PutManifest`]. When `missing` is empty
 /// the manifest was persisted successfully; otherwise the client must
 /// upload the listed chunks (typically via [`Method::PutChunk`]) and
@@ -313,6 +346,33 @@ pub struct PutManifestReply {
 ///
 /// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
 /// instance backs the accept loop plus any explicit dispatch calls.
+/// Recursive byte count under a directory. Used by the dashboard
+/// handler for reporting only; silent on read errors.
+fn dir_size_bytes(root: &std::path::Path) -> u64 {
+    let mut total: u64 = 0;
+    let mut stack = vec![root.to_path_buf()];
+    while let Some(dir) = stack.pop() {
+        let entries = match std::fs::read_dir(&dir) {
+            Ok(e) => e,
+            Err(_) => continue,
+        };
+        for entry in entries.flatten() {
+            let ft = match entry.file_type() {
+                Ok(t) => t,
+                Err(_) => continue,
+            };
+            if ft.is_dir() {
+                stack.push(entry.path());
+            } else if ft.is_file() {
+                if let Ok(m) = entry.metadata() {
+                    total = total.saturating_add(m.len());
+                }
+            }
+        }
+    }
+    total
+}
+
 pub struct RpcRouter {
     gossip: Arc<ClusterGossip>,
     blob_store: Option<Arc<BlobStore>>,
@@ -450,6 +510,70 @@ impl RpcRouter {
                 }
                 Ok(HandlerOutcome::Reply(json))
             }
+            Method::DashboardStatus => {
+                // Aggregator payload for dashboard-v2. Reuses gossip
+                // for name+zone+rustc; walks the on-disk stores for
+                // count + byte totals. Cheap: 5 filesystem walks.
+                let (blob_root, blob_count) = match &self.blob_store {
+                    Some(s) => {
+                        let ids = s.list_blob_ids().await.unwrap_or_default();
+                        (
+                            Some(s.root().display().to_string()),
+                            ids.len(),
+                        )
+                    }
+                    None => (None, 0),
+                };
+                let root_path = blob_root.as_ref().map(std::path::PathBuf::from);
+                let tag_count = match &self.tag_store {
+                    Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
+                    None => 0,
+                };
+                let ref_count = match &self.ref_store {
+                    Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
+                    None => 0,
+                };
+                // Snapshot + ref-tracking aren't held on the router;
+                // open by path derived from blob_store_root when
+                // possible.
+                let snapshot_count = match &root_path {
+                    Some(p) => match crate::cluster::snapshot::SnapshotStore::open(p.clone()) {
+                        Ok(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
+                        Err(_) => 0,
+                    },
+                    None => 0,
+                };
+                let ref_tracking_count = match &root_path {
+                    Some(p) => match crate::cluster::ref_tracking::RefTracking::open(p.clone()) {
+                        Ok(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
+                        Err(_) => 0,
+                    },
+                    None => 0,
+                };
+                let blob_store_bytes = match &root_path {
+                    Some(p) => dir_size_bytes(p),
+                    None => 0,
+                };
+                let rustc_release = self
+                    .gossip
+                    .self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
+                    .await;
+                let reply = DashboardStatusReply {
+                    node_name: self.local_name.clone(),
+                    zone: self.local_zone.clone(),
+                    blob_store_root: blob_root,
+                    blob_count,
+                    tag_count,
+                    ref_count,
+                    snapshot_count,
+                    ref_tracking_count,
+                    blob_store_bytes,
+                    rustc_release,
+                };
+                let json = serde_json::to_vec(&reply)
+                    .context("encoding DashboardStatusReply as JSON")?;
+                Ok(HandlerOutcome::Reply(json))
+            }
             Method::BlobStat => {
                 let store = match &self.blob_store {
                     Some(s) => s,

claw-store/src/cluster/rpc/client.rs

@@ -169,6 +169,19 @@ pub async fn call_peer_status(conn: &Connection) -> Result<PeerStatusReply> {
     serde_json::from_slice(&reply).context("decoding PeerStatusReply JSON")
 }
 
+/// Convenience wrapper for [`Method::DashboardStatus`]. Feeds the
+/// dashboard-v2 aggregator. Empty payload → JSON reply with counts
+/// + on-disk bytes.
+pub async fn call_dashboard_status(conn: &Connection) -> Result<DashboardStatusReply> {
+    let reply = rpc_call(conn, Method::DashboardStatus, &[]).await?;
+    if reply.len() == 1 {
+        if let Some(code) = decode_error(reply[0]) {
+            bail!("peer replied with error: {}", code.describe());
+        }
+    }
+    serde_json::from_slice(&reply).context("decoding DashboardStatusReply JSON")
+}
+
 /// Recognise a single-byte reply as one of our error codes. Returns
 /// `None` for any other single-byte value (which is a valid reply,
 /// just an unusually short one).

claw-store/src/serve.rs

@@ -706,26 +706,30 @@ pub fn build_app_with_v2(
         api = api.fallback_service(tower_http::services::ServeDir::new(dir));
     }
 
-    // dashboard-v2 backend (docs/dashboard-v2.md). Additive alongside
-    // the legacy /api/* routes above so the cutover doesn't break the
-    // old dashboard while the new one is being iterated on.
-    let v2_state = std::sync::Arc::new(crate::serve_v2::V2State::from_config(&state.cfg));
-    let mut v2_routes = crate::serve_v2::routes();
+    // dashboard-v2 aggregator backend (docs/dashboard-v2.md).
+    // Builds only when [cluster] + [cluster.tls] + peers are
+    // present; otherwise v2 API routes are skipped, and the /v2/
+    // static mount (if any) still works so operators can see the
+    // config-missing error message the SPA renders.
+    let v2_state = crate::serve_v2::V2State::from_config(&state.cfg)
+        .map(std::sync::Arc::new)
+        .ok();
+    let mut v2_router: Router<()> = Router::new();
+    if let Some(v2s) = v2_state {
+        v2_router = v2_router.merge(crate::serve_v2::routes().with_state(v2s));
+    }
     if let Some(dir) = v2_static_dir {
-        // Serve the compiled SPA at /v2/*. Any unknown /v2/* path
-        // falls back to index.html so client-side wouter routing works.
-        v2_routes = v2_routes.nest_service(
+        v2_router = v2_router.nest_service(
             "/v2",
             tower_http::services::ServeDir::new(&dir).fallback(
                 tower_http::services::ServeFile::new(dir.join("index.html")),
             ),
         );
     }
-    let v2_routes = v2_routes.with_state(v2_state);
 
     Router::new()
         .merge(api.layer(cors.clone()).with_state(state))
-        .merge(v2_routes.layer(cors))
+        .merge(v2_router.layer(cors))
 }
 
 pub async fn run_server(

claw-store/src/serve_v2.rs

@@ -1,561 +1,236 @@
-//! dashboard-v2 backend — additive `/api/v2/*` handlers.
+//! dashboard-v2 aggregator backend — additive `/api/v2/*` handlers.
 //!
-//! Reads the distributed-architecture stores (BlobStore, TagStore,
-//! RefStore, SnapshotStore, RefTracking). Kept in a separate file
-//! from `serve.rs` so the legacy ZFS-era handlers there can be
-//! removed cleanly at cutover without a big rebase.
+//! Fleet-single-pane-of-glass model: `claw-store serve` runs on the
+//! operator's laptop (typically). It holds a fleet-CA leaf cert +
+//! the peer list, connects to each daemon's cluster RPC port over
+//! mTLS, and issues `DashboardStatus` RPCs. Peers don't need to
+//! run any HTTP server of their own — RPC over the existing QUIC
+//! socket is enough.
 //!
-//! Design + endpoint spec: `docs/dashboard-v2.md`.
+//! Config: peers come from `[[cluster.peers]]` in the aggregator's
+//! `config.toml`. TLS material via `[cluster.tls]`.
 //!
-//! Single-node reads in this cut. Cross-node fleet aggregation
-//! (fan-out via QUIC RPC) lands in a follow-on PR.
+//! Design doc: `docs/dashboard-v2.md`.
 
 use axum::{
-    extract::{Path, Query, State},
+    extract::{Path, State},
     routing::get,
     Json, Router,
 };
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
+use std::net::SocketAddr;
 use std::sync::Arc;
+use std::time::Duration;
 
-use crate::cluster::blob::{BlobId, BlobStore};
-use crate::cluster::ref_tracking::{RefEntry, RefTracking};
-use crate::cluster::refs::RefStore;
-use crate::cluster::snapshot::SnapshotStore;
-use crate::cluster::tags::TagStore;
-use crate::config::Config;
+use crate::cluster::rpc::{call_dashboard_status, DashboardStatusReply};
+use crate::cluster::transport::{NodeIdentity, QuicClient};
+use crate::config::{Config, PeerEntry};
 
-/// Runtime shared by all v2 handlers. Cheap clones — inner types
-/// are `Arc<_>` or filesystem-backed with no in-memory state.
-#[derive(Clone)]
+/// Aggregator runtime: one QuicClient, one peer list, one identity.
+///
+/// The client is reused across every RPC (quinn holds one UDP
+/// endpoint + a pool of connections). Peer addresses are resolved
+/// per-request so a hot-swapped config doesn't require restart.
 pub struct V2State {
-    pub node_name: String,
-    pub blob_store_root: Option<PathBuf>,
+    pub aggregator_name: String,
+    pub peers: Vec<PeerEntry>,
+    pub client: QuicClient,
+    /// Fallback RPC port when a peer entry has `lan_addr = "10.x:7701"`
+    /// (gossip port, not RPC). We assume RPC = gossip + 1 in that
+    /// case unless the operator sets an explicit `rpc_addr`.
+    pub default_rpc_port_offset: u16,
 }
 
 impl V2State {
-    pub fn from_config(cfg: &Config) -> Self {
-        let blob_store_root = cfg
+    /// Build an aggregator state from the aggregator's own config.
+    /// Requires `[cluster]` + `[cluster.tls]` + at least one
+    /// `[[cluster.peers]]`. The aggregator doesn't need a
+    /// blob_store_root itself.
+    pub fn from_config(cfg: &Config) -> anyhow::Result<Self> {
+        let cluster = cfg
             .cluster
             .as_ref()
-            .and_then(|c| c.blob_store_root.clone());
-        Self {
-            node_name: cfg.node.name.clone(),
-            blob_store_root,
+            .ok_or_else(|| anyhow::anyhow!("dashboard-v2 aggregator needs [cluster] config"))?;
+        let identity = NodeIdentity::from_cluster_config(cluster)
+            .map_err(|e| anyhow::anyhow!("loading fleet-CA identity: {e}"))?;
+        let client = QuicClient::new("0.0.0.0:0".parse()?, identity)
+            .map_err(|e| anyhow::anyhow!("building QUIC client: {e}"))?;
+        Ok(Self {
+            aggregator_name: cfg.node.name.clone(),
+            peers: cluster.peers.clone(),
+            client,
+            default_rpc_port_offset: 1,
+        })
+    }
+
+    /// Resolve a peer name → RPC socket. Prefers `rpc_addr` when the
+    /// peer entry sets one; otherwise takes `lan_addr` and adds the
+    /// port offset (gossip → RPC = +1 by fleet convention).
+    fn peer_rpc_addr(&self, peer: &PeerEntry) -> Option<SocketAddr> {
+        // Prefer LAN (fast), fall back to tailnet. RPC port =
+        // gossip port + 1 by fleet convention (see fleet configs).
+        if let Some(lan) = peer.lan_addr {
+            return Some(SocketAddr::new(
+                lan.ip(),
+                lan.port() + self.default_rpc_port_offset,
+            ));
         }
+        if let Some(ts) = peer.tailscale_addr {
+            return Some(SocketAddr::new(
+                ts.ip(),
+                ts.port() + self.default_rpc_port_offset,
+            ));
+        }
+        None
     }
 
-    fn open_blob(&self) -> Option<BlobStore> {
-        let root = self.blob_store_root.as_ref()?;
-        BlobStore::open(root.clone()).ok()
-    }
-
-    fn open_tags(&self) -> Option<TagStore> {
-        let root = self.blob_store_root.as_ref()?;
-        TagStore::open(root.join("tags-db")).ok()
-    }
-
-    fn open_refs(&self) -> Option<RefStore> {
-        let root = self.blob_store_root.as_ref()?;
-        RefStore::open(root.join("refs-db")).ok()
-    }
-
-    fn open_snapshots(&self) -> Option<SnapshotStore> {
-        let root = self.blob_store_root.as_ref()?;
-        SnapshotStore::open(root.clone()).ok()
-    }
-
-    fn open_ref_tracking(&self) -> Option<RefTracking> {
-        let root = self.blob_store_root.as_ref()?;
-        RefTracking::open(root.clone()).ok()
+    /// Fetch a single node's dashboard payload via RPC.
+    async fn fetch_node(&self, peer: &PeerEntry) -> anyhow::Result<DashboardStatusReply> {
+        let addr = self.peer_rpc_addr(peer).ok_or_else(|| {
+            anyhow::anyhow!("peer {} has no reachable RPC address", peer.name)
+        })?;
+        let conn = tokio::time::timeout(
+            Duration::from_secs(5),
+            self.client.connect(addr, &peer.name),
+        )
+        .await
+        .map_err(|_| anyhow::anyhow!("connect to {} timed out", peer.name))??;
+        let reply = call_dashboard_status(&conn).await?;
+        conn.close(quinn::VarInt::from_u32(0), b"done");
+        Ok(reply)
     }
 }
 
-// ── request / response shapes ────────────────────────────────────
+// ── response shapes ──────────────────────────────────────────────
 
 #[derive(Serialize, Deserialize)]
 pub struct NodeStatusV2 {
     pub node_name: String,
-    pub blob_store_root: Option<PathBuf>,
+    pub zone: String,
+    pub blob_store_root: Option<String>,
     pub blob_count: usize,
     pub tag_count: usize,
     pub ref_count: usize,
     pub snapshot_count: usize,
     pub ref_tracking_count: usize,
     pub blob_store_bytes: u64,
+    pub rustc_release: Option<String>,
+    /// `true` when the aggregator successfully talked to the peer;
+    /// `false` when the RPC failed. Frontend uses this to badge the
+    /// card as offline.
+    pub online: bool,
+    /// Human-readable error when `online = false`.
+    pub error: Option<String>,
 }
 
-#[derive(Deserialize)]
-pub struct Pagination {
-    #[serde(default = "default_limit")]
-    pub limit: usize,
-    #[serde(default)]
-    pub offset: usize,
-}
-
-fn default_limit() -> usize {
-    200
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct BlobSummary {
-    pub blob_id_hex: String,
-    pub size_bytes: u64,
-    pub chunk_count: usize,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct TagSummaryV2 {
-    pub key: String,
-    pub value_hex: String,
-}
-
-#[derive(Deserialize)]
-pub struct TagFilter {
-    #[serde(default)]
-    pub prefix: String,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct RefSummary {
-    pub fingerprint_hex: String,
-    pub blob_id_hex: String,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct SnapshotSummaryV2 {
-    pub name: String,
-    pub created_at_unix: u64,
-    pub blob_count: usize,
-    pub file_bytes: u64,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct RefTrackingItem {
-    pub fingerprint_hex: String,
-    pub repo: String,
-    pub refs: Vec<String>,
-    pub first_seen_unix: u64,
-    pub last_seen_unix: u64,
-}
-
-impl From<RefEntry> for RefTrackingItem {
-    fn from(e: RefEntry) -> Self {
-        let mut hex = String::with_capacity(64);
-        for b in &e.fingerprint {
-            hex.push_str(&format!("{b:02x}"));
+impl NodeStatusV2 {
+    fn ok(name: &str, r: DashboardStatusReply) -> Self {
+        let mut n = Self {
+            node_name: r.node_name,
+            zone: r.zone,
+            blob_store_root: r.blob_store_root,
+            blob_count: r.blob_count,
+            tag_count: r.tag_count,
+            ref_count: r.ref_count,
+            snapshot_count: r.snapshot_count,
+            ref_tracking_count: r.ref_tracking_count,
+            blob_store_bytes: r.blob_store_bytes,
+            rustc_release: r.rustc_release,
+            online: true,
+            error: None,
+        };
+        if n.node_name.is_empty() {
+            n.node_name = name.to_string();
         }
+        n
+    }
+    fn err(name: &str, e: String) -> Self {
         Self {
-            fingerprint_hex: hex,
-            repo: e.repo,
-            refs: e.refs,
-            first_seen_unix: e.first_seen_unix,
-            last_seen_unix: e.last_seen_unix,
+            node_name: name.to_string(),
+            zone: String::new(),
+            blob_store_root: None,
+            blob_count: 0,
+            tag_count: 0,
+            ref_count: 0,
+            snapshot_count: 0,
+            ref_tracking_count: 0,
+            blob_store_bytes: 0,
+            rustc_release: None,
+            online: false,
+            error: Some(e),
         }
     }
 }
 
-#[derive(Deserialize)]
-pub struct RefTrackingFilter {
-    #[serde(default)]
-    pub repo: String,
-}
-
 #[derive(Serialize)]
-pub struct EmptyReason {
-    pub reason: String,
+pub struct FleetSnapshot {
+    pub aggregator_name: String,
+    pub fetched_at_unix: u64,
+    pub nodes: Vec<NodeStatusV2>,
 }
 
 // ── handlers ─────────────────────────────────────────────────────
 
-async fn handle_status(State(s): State<Arc<V2State>>) -> Json<NodeStatusV2> {
-    let blob = s.open_blob();
-    let tags = s.open_tags();
-    let refs = s.open_refs();
-    let snaps = s.open_snapshots();
-    let rt = s.open_ref_tracking();
-    let root = s.blob_store_root.clone();
-
-    let blob_count = match &blob {
-        Some(b) => b.list_blob_ids().await.map(|v| v.len()).unwrap_or(0),
-        None => 0,
-    };
-    let tag_count = match &tags {
-        Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
-        None => 0,
-    };
-    let ref_count = match &refs {
-        Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
-        None => 0,
-    };
-    let snapshot_count = match &snaps {
-        Some(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
-        None => 0,
-    };
-    let ref_tracking_count = match &rt {
-        Some(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
-        None => 0,
-    };
-    let blob_store_bytes = match &root {
-        Some(p) => dir_size_bytes(p),
-        None => 0,
-    };
-    Json(NodeStatusV2 {
-        node_name: s.node_name.clone(),
-        blob_store_root: root,
-        blob_count,
-        tag_count,
-        ref_count,
-        snapshot_count,
-        ref_tracking_count,
-        blob_store_bytes,
+async fn handle_fleet(State(s): State<Arc<V2State>>) -> Json<FleetSnapshot> {
+    // Fan out concurrently.
+    // Fan out with a JoinSet — bounded concurrency, no futures crate.
+    let mut set = tokio::task::JoinSet::new();
+    for p in &s.peers {
+        let name = p.name.clone();
+        let state = s.clone();
+        let peer = p.clone();
+        set.spawn(async move {
+            match state.fetch_node(&peer).await {
+                Ok(r) => NodeStatusV2::ok(&name, r),
+                Err(e) => NodeStatusV2::err(&name, format!("{e:#}")),
+            }
+        });
+    }
+    let mut nodes: Vec<NodeStatusV2> = Vec::new();
+    while let Some(res) = set.join_next().await {
+        if let Ok(n) = res {
+            nodes.push(n);
+        }
+    }
+    // Deterministic ordering — by name — so the UI doesn't shuffle
+    // between polls.
+    nodes.sort_by(|a, b| a.node_name.cmp(&b.node_name));
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+    Json(FleetSnapshot {
+        aggregator_name: s.aggregator_name.clone(),
+        fetched_at_unix: now,
+        nodes,
     })
 }
 
 async fn handle_node_status(
     State(s): State<Arc<V2State>>,
     Path(name): Path<String>,
-) -> Result<Json<NodeStatusV2>, (axum::http::StatusCode, Json<EmptyReason>)> {
-    // Single-node cut: only "local" or "<own name>" resolves.
-    // Cross-node lookup lands with the fleet fan-out follow-on.
-    if name != "local" && name != s.node_name {
-        return Err((
-            axum::http::StatusCode::NOT_IMPLEMENTED,
-            Json(EmptyReason {
-                reason: format!(
-                    "cross-node lookup not yet implemented; requested {name}, this is {}",
-                    s.node_name
-                ),
-            }),
-        ));
-    }
-    Ok(handle_status(State(s)).await)
-}
-
-async fn handle_blobs(
-    State(s): State<Arc<V2State>>,
-    Query(p): Query<Pagination>,
-) -> Json<Vec<BlobSummary>> {
-    let store = match s.open_blob() {
-        Some(b) => b,
-        None => return Json(Vec::new()),
-    };
-    let mut ids = store.list_blob_ids().await.unwrap_or_default();
-    ids.sort();
-    let store = std::sync::Arc::new(store);
-    let out = ids.into_iter().skip(p.offset).take(p.limit).map(|id| {
-        let store = store.clone();
-        async move {
-            let manifest = store.load_manifest(&id).await.ok().flatten();
-            BlobSummary {
-                blob_id_hex: id.to_hex(),
-                size_bytes: manifest.as_ref().map(|m| m.total_size).unwrap_or(0),
-                chunk_count: manifest.map(|m| m.chunks.len()).unwrap_or(0),
-            }
+) -> Json<NodeStatusV2> {
+    let peer = s.peers.iter().find(|p| p.name == name).cloned();
+    let peer = match peer {
+        Some(p) => p,
+        None => {
+            return Json(NodeStatusV2::err(
+                &name,
+                format!("no peer named {name} in aggregator config"),
+            ))
         }
-    });
-    let collected: Vec<BlobSummary> = futures_join(out).await;
-    Json(collected)
-}
-
-async fn handle_tags(
-    State(s): State<Arc<V2State>>,
-    Query(f): Query<TagFilter>,
-) -> Json<Vec<TagSummaryV2>> {
-    let store = match s.open_tags() {
-        Some(t) => t,
-        None => return Json(Vec::new()),
     };
-    let entries = store.list().await.unwrap_or_default();
-    let out: Vec<_> = entries
-        .into_iter()
-        .filter(|e| f.prefix.is_empty() || e.key.starts_with(&f.prefix))
-        .map(|e| TagSummaryV2 {
-            key: e.key,
-            value_hex: e.value_hex,
-        })
-        .collect();
-    Json(out)
-}
-
-async fn handle_refs(
-    State(s): State<Arc<V2State>>,
-    Query(p): Query<Pagination>,
-) -> Json<Vec<RefSummary>> {
-    let store = match s.open_refs() {
-        Some(r) => r,
-        None => return Json(Vec::new()),
+    let node = match s.fetch_node(&peer).await {
+        Ok(r) => NodeStatusV2::ok(&name, r),
+        Err(e) => NodeStatusV2::err(&name, format!("{e:#}")),
     };
-    let all = store.list().await.unwrap_or_default();
-    let out: Vec<_> = all
-        .into_iter()
-        .skip(p.offset)
-        .take(p.limit)
-        .map(|(k, v)| {
-            let mut kh = String::with_capacity(64);
-            for b in &k {
-                kh.push_str(&format!("{b:02x}"));
-            }
-            RefSummary {
-                fingerprint_hex: kh,
-                blob_id_hex: BlobId::from_bytes(v).to_hex(),
-            }
-        })
-        .collect();
-    Json(out)
-}
-
-async fn handle_snapshots(State(s): State<Arc<V2State>>) -> Json<Vec<SnapshotSummaryV2>> {
-    let store = match s.open_snapshots() {
-        Some(s) => s,
-        None => return Json(Vec::new()),
-    };
-    let entries = store.list().await.unwrap_or_default();
-    let out: Vec<_> = entries
-        .into_iter()
-        .map(|s| SnapshotSummaryV2 {
-            name: s.name,
-            created_at_unix: s.created_at_unix,
-            blob_count: s.blob_count,
-            file_bytes: s.file_bytes,
-        })
-        .collect();
-    Json(out)
-}
-
-async fn handle_ref_tracking(
-    State(s): State<Arc<V2State>>,
-    Query(f): Query<RefTrackingFilter>,
-) -> Json<Vec<RefTrackingItem>> {
-    let store = match s.open_ref_tracking() {
-        Some(r) => r,
-        None => return Json(Vec::new()),
-    };
-    let all = store.list_all().await.unwrap_or_default();
-    let out: Vec<_> = all
-        .into_iter()
-        .filter(|e| f.repo.is_empty() || e.repo == f.repo)
-        .map(RefTrackingItem::from)
-        .collect();
-    Json(out)
-}
-
-// ── helpers ──────────────────────────────────────────────────────
-
-/// Recursive byte count. Silent on read errors — used only for
-/// reporting, not correctness.
-fn dir_size_bytes(root: &std::path::Path) -> u64 {
-    let mut total: u64 = 0;
-    let mut stack = vec![root.to_path_buf()];
-    while let Some(dir) = stack.pop() {
-        let entries = match std::fs::read_dir(&dir) {
-            Ok(e) => e,
-            Err(_) => continue,
-        };
-        for entry in entries.flatten() {
-            let ft = match entry.file_type() {
-                Ok(t) => t,
-                Err(_) => continue,
-            };
-            if ft.is_dir() {
-                stack.push(entry.path());
-            } else if ft.is_file() {
-                if let Ok(m) = entry.metadata() {
-                    total = total.saturating_add(m.len());
-                }
-            }
-        }
-    }
-    total
-}
-
-/// Sequentially resolve N per-item async closures. Kept simple —
-/// blob-metadata reads are I/O light and page-sized (200 default);
-/// concurrent-fan-out isn't worth the complexity here.
-async fn futures_join<F, T>(iter: impl Iterator<Item = F>) -> Vec<T>
-where
-    F: std::future::Future<Output = T>,
-{
-    let mut out = Vec::new();
-    for f in iter {
-        out.push(f.await);
-    }
-    out
+    Json(node)
 }
 
 // ── route registration ──────────────────────────────────────────
 
-/// Attach the v2 route tree onto an existing axum Router.
-///
-/// Caller supplies the v2 state via `.with_state(state)` when
-/// building the final router.
 pub fn routes() -> Router<Arc<V2State>> {
     Router::new()
-        .route("/api/v2/node/local/status", get(handle_status))
+        .route("/api/v2/fleet", get(handle_fleet))
         .route("/api/v2/node/:name/status", get(handle_node_status))
-        .route("/api/v2/storage/blobs", get(handle_blobs))
-        .route("/api/v2/storage/tags", get(handle_tags))
-        .route("/api/v2/storage/refs", get(handle_refs))
-        .route("/api/v2/storage/snapshots", get(handle_snapshots))
-        .route("/api/v2/storage/ref-tracking", get(handle_ref_tracking))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use axum::body::to_bytes;
-    use axum::{body::Body, http::Request};
-    use tempfile::TempDir;
-    use tower::ServiceExt;
-
-    async fn seed_state() -> (TempDir, Arc<V2State>) {
-        let tmp = TempDir::new().unwrap();
-        let root = tmp.path().to_path_buf();
-        // Seed one blob so /status has non-zero numbers.
-        let store = BlobStore::open(root.clone()).unwrap();
-        store.put_bytes(b"hello v2").await.unwrap();
-        let snap = SnapshotStore::open(root.clone()).unwrap();
-        snap.create("seed", &store, 42).await.unwrap();
-        let state = Arc::new(V2State {
-            node_name: "test-node".into(),
-            blob_store_root: Some(root),
-        });
-        (tmp, state)
-    }
-
-    #[tokio::test]
-    async fn status_reports_seeded_blob_and_snapshot() {
-        let (_tmp, state) = seed_state().await;
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/node/local/status")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        assert_eq!(resp.status(), axum::http::StatusCode::OK);
-        let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
-        let body: NodeStatusV2 = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(body.node_name, "test-node");
-        assert_eq!(body.blob_count, 1);
-        assert_eq!(body.snapshot_count, 1);
-        assert!(body.blob_store_bytes > 0);
-    }
-
-    #[tokio::test]
-    async fn cross_node_status_returns_not_implemented() {
-        let (_tmp, state) = seed_state().await;
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/node/architect/status")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        assert_eq!(resp.status(), axum::http::StatusCode::NOT_IMPLEMENTED);
-    }
-
-    #[tokio::test]
-    async fn blobs_endpoint_paginates() {
-        let (_tmp, state) = seed_state().await;
-        // Seed more blobs.
-        if let Some(root) = &state.blob_store_root {
-            let store = BlobStore::open(root.clone()).unwrap();
-            for i in 0..5 {
-                store
-                    .put_bytes(format!("payload-{i}").as_bytes())
-                    .await
-                    .unwrap();
-            }
-        }
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/storage/blobs?limit=2&offset=1")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        assert_eq!(resp.status(), axum::http::StatusCode::OK);
-        let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
-        let body: Vec<BlobSummary> = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(body.len(), 2);
-    }
-
-    #[tokio::test]
-    async fn tags_endpoint_filters_by_prefix() {
-        let (_tmp, state) = seed_state().await;
-        if let Some(root) = &state.blob_store_root {
-            let ts = TagStore::open(root.join("tags-db")).unwrap();
-            ts.put("clawverse:main:latest", &[0xAA; 32]).await.unwrap();
-            ts.put("other:tag", &[0xBB; 32]).await.unwrap();
-        }
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/storage/tags?prefix=clawverse:")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
-        let body: Vec<TagSummaryV2> = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(body.len(), 1);
-        assert_eq!(body[0].key, "clawverse:main:latest");
-    }
-
-    #[tokio::test]
-    async fn snapshots_endpoint_returns_seeded() {
-        let (_tmp, state) = seed_state().await;
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/storage/snapshots")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
-        let body: Vec<SnapshotSummaryV2> = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(body.len(), 1);
-        assert_eq!(body[0].name, "seed");
-        assert_eq!(body[0].created_at_unix, 42);
-    }
-
-    #[tokio::test]
-    async fn ref_tracking_filters_by_repo() {
-        let (_tmp, state) = seed_state().await;
-        if let Some(root) = &state.blob_store_root {
-            let rt = RefTracking::open(root.clone()).unwrap();
-            rt.record([0x11; 32], "r/x", "main", 100).await.unwrap();
-            rt.record([0x22; 32], "r/y", "main", 100).await.unwrap();
-        }
-        let app: Router = routes().with_state(state);
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .uri("/api/v2/storage/ref-tracking?repo=r/x")
-                    .body(Body::empty())
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-        let bytes = to_bytes(resp.into_body(), 65_536).await.unwrap();
-        let body: Vec<RefTrackingItem> = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(body.len(), 1);
-        assert_eq!(body[0].repo, "r/x");
-    }
 }