feat(maestro): daemon CatalogMirror + own∪peer HTTP responses (NAP-246)
New module crates/maestro-daemon/src/catalog_mirror.rs: in-memory mirror of the global catalog (peer daemons + their sessions), updated from the `DaemonInbound::CatalogEvent` frames the registry now fans out. `local_api` consumes it: `GET /api/sessions` returns the union of the daemon's own sessions and the mirrored peers; `GET /api/daemons` returns self + connected peers. The web UI sees the same catalog from any home daemon. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -11,8 +11,8 @@ use std::time::Duration;
|
||||
|
||||
use common::DaemonFixture;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use maestro_daemon::local_api;
|
||||
use maestro_protocol::{CreateSessionRequest, DaemonInfo, Session};
|
||||
use maestro_daemon::{local_api, CatalogMirror};
|
||||
use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||
@@ -24,6 +24,7 @@ type WsClient = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
|
||||
struct Harness {
|
||||
addr: SocketAddr,
|
||||
manager: maestro_daemon::SessionManager,
|
||||
catalog: CatalogMirror,
|
||||
_shutdown: oneshot::Sender<()>,
|
||||
_fx: DaemonFixture,
|
||||
}
|
||||
@@ -31,7 +32,8 @@ struct Harness {
|
||||
async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
|
||||
let fx = DaemonFixture::new(daemon_id);
|
||||
let manager = fx.manager.clone();
|
||||
let state = local_api::state_for(manager.clone(), hostname, daemon_url);
|
||||
let catalog = CatalogMirror::new();
|
||||
let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone());
|
||||
let router = local_api::router(state).merge(local_api::broker_router(manager.clone()));
|
||||
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
|
||||
.await
|
||||
@@ -48,6 +50,7 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
|
||||
Harness {
|
||||
addr,
|
||||
manager,
|
||||
catalog,
|
||||
_shutdown: tx,
|
||||
_fx: fx,
|
||||
}
|
||||
@@ -73,17 +76,78 @@ async fn web_info_returns_daemon_identity() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_daemons_returns_self_only() {
|
||||
async fn list_daemons_includes_self_and_mirrored_peers() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
|
||||
// Before any catalog events arrive, the daemon only knows about itself.
|
||||
let list: Vec<DaemonInfo> = reqwest::get(url(h.addr, "/api/daemons"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
// v1: no peer mirror, so a daemon only knows about itself.
|
||||
assert_eq!(list.len(), 1);
|
||||
assert_eq!(list[0].daemon_id, "laptop");
|
||||
|
||||
// Simulate the registry fanning out a peer addition.
|
||||
h.catalog
|
||||
.apply(RegistryEvent::DaemonAdded {
|
||||
daemon: DaemonInfo {
|
||||
daemon_id: "konishi".into(),
|
||||
hostname: "konishi-host".into(),
|
||||
version: "0.1.0".into(),
|
||||
daemon_url: "ws://10.0.0.2:7070".into(),
|
||||
},
|
||||
})
|
||||
.await;
|
||||
|
||||
let list: Vec<DaemonInfo> = reqwest::get(url(h.addr, "/api/daemons"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
let ids: Vec<&str> = list.iter().map(|d| d.daemon_id.as_str()).collect();
|
||||
assert!(ids.contains(&"laptop"));
|
||||
assert!(ids.contains(&"konishi"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_sessions_unions_own_and_peer() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
// Create one local session.
|
||||
let own = h
|
||||
.manager
|
||||
.create(CreateSessionRequest {
|
||||
daemon_id: "laptop".into(),
|
||||
cwd: "/tmp".into(),
|
||||
command: "sleep 60".into(),
|
||||
env: HashMap::new(),
|
||||
metadata: HashMap::new(),
|
||||
worktree: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
// Inject a peer session via the mirror.
|
||||
let mut peer = own.clone();
|
||||
peer.id = Uuid::new_v4();
|
||||
peer.daemon_id = "konishi".into();
|
||||
h.catalog
|
||||
.apply(RegistryEvent::SessionAdded {
|
||||
session: peer.clone(),
|
||||
})
|
||||
.await;
|
||||
|
||||
let list: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
let ids: Vec<Uuid> = list.iter().map(|s| s.id).collect();
|
||||
assert!(ids.contains(&own.id), "own session missing: {ids:?}");
|
||||
assert!(ids.contains(&peer.id), "peer session missing: {ids:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user