naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(maestro): drop registry, p2p, and auth from daemon core

Browse Source 
Maestro is now single-host: one daemon owns the machine's sessions and
clients talk to it directly. Removes the registry crate, the daemon's
registry client/broker/catalog mirror, JWT attach auth, cookie auth, and
the perimeter (T2) transport config.

The daemon API is now unauthenticated on its bind address, same trust
model as tmux. Session and CreateSessionRequest lose daemon_id;
RegistryEvent becomes CatalogEvent with no daemon variants; attach moves
to WS /api/attach/:id with no token.
This commit is contained in:
Miguel Palhas
2026-08-22 10:00:24 +01:00
parent d6d433770b
commit d006f4a7bd
40 changed files with 314 additions and 9294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/tests/local_api.rs
+39 -181
View File
@@ -1,7 +1,4 @@
//! Integration tests for the daemon's local HTTP API + broker WS (NAP-243).
//!
//! The legacy `/sessions/:id/stream` endpoint is covered by `tests/attach.rs`;
//! this file exercises the new `/api/*` surface and the same-host broker WS.
//! Integration tests for the daemon's HTTP + WS API (NAP-243).
mod common;
@@ -11,8 +8,8 @@ use std::time::Duration;
use common::DaemonFixture;
use futures_util::{SinkExt, StreamExt};
use maestro_daemon::{local_api, BrokerHandle, CatalogMirror, LocalEventBus};
use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session};
use maestro_daemon::{local_api, LocalEventBus};
use maestro_protocol::{CreateSessionRequest, Session};
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use tokio_tungstenite::tungstenite::Message as WsMessage;
@@ -24,42 +21,24 @@ type WsClient = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
struct Harness {
addr: SocketAddr,
manager: maestro_daemon::SessionManager,
catalog: CatalogMirror,
_shutdown: oneshot::Sender<()>,
_fx: DaemonFixture,
}
async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
let fx = DaemonFixture::new(daemon_id);
async fn spawn(hostname: &str) -> Harness {
let fx = DaemonFixture::new();
let manager = fx.manager.clone();
let catalog = CatalogMirror::new();
let events = LocalEventBus::new();
// NAP-247/NAP-327: tests don't exercise the cross-host broker path; hand the
// routers an isolated BrokerHandle whose receiver is dropped, so any
// resolve_attach / forward_create / forward_kill fails fast with
// NoControlLink (the handle starts disconnected).
let (broker, _broker_rx) = BrokerHandle::new();
// Built-in presets (claude/opencode/zsh + custom) so web_info advertises a
// known set (#26).
// Built-in presets (claude/opencode/zsh + custom) so `/api/info` advertises
// a known set (#26).
let presets = maestro_config::SpawnSection::default().resolve().unwrap();
let state = local_api::state_for(
let router = local_api::router(local_api::state_for(
manager.clone(),
hostname,
daemon_url,
"/home/test",
presets,
catalog.clone(),
events,
broker.clone(),
);
let broker_state = local_api::BrokerState {
manager: manager.clone(),
catalog: catalog.clone(),
broker,
self_daemon_id: std::sync::Arc::new(daemon_id.into()),
self_perimeter_id: None,
};
let router = local_api::router(state).merge(local_api::broker_router(broker_state));
));
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
.await
.unwrap();
@@ -75,7 +54,6 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
Harness {
addr,
manager,
catalog,
_shutdown: tx,
_fx: fx,
}
@@ -86,99 +64,22 @@ fn url(addr: SocketAddr, path: &str) -> String {
}
#[tokio::test]
async fn web_info_returns_daemon_identity() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/web/info"))
async fn info_returns_daemon_identity() {
let h = spawn("naps-laptop").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
.await
.unwrap()
.json()
.await
.unwrap();
assert_eq!(resp["daemon_id"], "laptop");
assert_eq!(resp["hostname"], "naps-laptop");
assert_eq!(resp["daemon_url"], "ws://10.0.0.1:7070");
assert_eq!(resp["home_dir"], "/home/test");
assert!(resp["version"].is_string());
}
#[tokio::test]
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();
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(),
home_dir: String::new(),
},
})
.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]
async fn list_sessions_starts_empty_and_reflects_creates() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let h = spawn("naps-laptop").await;
let empty: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
.await
@@ -189,8 +90,7 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
assert!(empty.is_empty());
let req = CreateSessionRequest {
daemon_id: "laptop".into(),
cwd: h.manager.daemon_id().into(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
metadata: HashMap::new(),
@@ -205,7 +105,6 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
.json()
.await
.unwrap();
assert_eq!(created.daemon_id, "laptop");
let after: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
.await
@@ -218,37 +117,11 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
}
#[tokio::test]
async fn post_to_peer_daemon_id_brokers_and_503s_without_registry_link() {
// Cross-daemon create now brokers to the owner over the registry control
// link (NAP-327). This harness has no registry link (the broker starts
// disconnected), so the daemon fails fast with a retryable 503 instead of
// the old 501 — proving it attempts to broker rather than refusing.
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let req = CreateSessionRequest {
daemon_id: "konishi".into(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
};
let resp = reqwest::Client::new()
.post(url(h.addr, "/api/sessions"))
.json(&req)
.send()
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
let body: serde_json::Value = resp.json().await.unwrap();
assert_eq!(body["error"]["code"], "no_registry_link");
}
#[tokio::test]
async fn web_info_advertises_spawn_presets() {
// #26: the connected daemon's presets ride the web_info payload so the
// spawn form needs no extra round-trip. Built-ins + the implicit custom.
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/web/info"))
async fn info_advertises_spawn_presets() {
// #26: the daemon's presets ride the info payload so the spawn form needs
// no extra round-trip. Built-ins + the implicit custom.
let h = spawn("naps-laptop").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/info"))
.await
.unwrap()
.json()
@@ -264,11 +137,11 @@ async fn web_info_advertises_spawn_presets() {
}
#[tokio::test]
async fn recent_dirs_self_returns_empty_without_history() {
// #26: the self-path reads this daemon's spawn history. The test fixture
// wires no history, so the list is empty but the endpoint is well-formed.
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/daemons/laptop/recent-dirs"))
async fn recent_dirs_returns_empty_without_history() {
// #26: reads this daemon's spawn history. The test fixture wires no
// history, so the list is empty but the endpoint is well-formed.
let h = spawn("naps-laptop").await;
let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/recent-dirs"))
.await
.unwrap()
.json()
@@ -279,15 +152,15 @@ async fn recent_dirs_self_returns_empty_without_history() {
}
#[tokio::test]
async fn gitroot_self_probes_path() {
// #26: the self-path resolves the query against the daemon home dir and
// walks up for a `.git`. A `/tmp`-rooted temp dir is not in a repo → null.
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
async fn gitroot_probes_path() {
// #26: resolves the query against the daemon home dir and walks up for a
// `.git`. A `/tmp`-rooted temp dir is not in a repo, so the root is null.
let h = spawn("naps-laptop").await;
let tmp = tempfile::tempdir().unwrap();
let resp: serde_json::Value = reqwest::get(url(
h.addr,
&format!(
"/api/daemons/laptop/gitroot?path={}",
"/api/gitroot?path={}",
urlencoding::encode(&tmp.path().to_string_lossy())
),
))
@@ -299,22 +172,9 @@ async fn gitroot_self_probes_path() {
assert!(resp.get("root").is_none() || resp["root"].is_null());
}
#[tokio::test]
async fn recent_dirs_for_peer_503s_without_registry_link() {
// #26: a peer target brokers over the (absent) registry link, failing fast
// with a retryable 503 — same shape as the cross-daemon create path.
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let resp = reqwest::get(url(h.addr, "/api/daemons/konishi/recent-dirs"))
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
let body: serde_json::Value = resp.json().await.unwrap();
assert_eq!(body["error"]["code"], "no_registry_link");
}
#[tokio::test]
async fn delete_unknown_session_returns_404() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let h = spawn("naps-laptop").await;
let resp = reqwest::Client::new()
.delete(url(h.addr, &format!("/api/sessions/{}", Uuid::new_v4())))
.send()
@@ -325,11 +185,10 @@ async fn delete_unknown_session_returns_404() {
#[tokio::test]
async fn delete_owned_session_kills_it() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
let h = spawn("naps-laptop").await;
let session = h
.manager
.create(CreateSessionRequest {
daemon_id: "laptop".into(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -348,12 +207,11 @@ async fn delete_owned_session_kills_it() {
}
#[tokio::test]
async fn broker_ws_attaches_to_local_session() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
async fn attach_ws_streams_pty_bytes() {
let h = spawn("naps-laptop").await;
let session = h
.manager
.create(CreateSessionRequest {
daemon_id: "laptop".into(),
cwd: "/tmp".into(),
command: "bash --noprofile --norc".into(),
env: HashMap::new(),
@@ -367,7 +225,7 @@ async fn broker_ws_attaches_to_local_session() {
let (mut ws, _): (WsClient, _) = connect_async(ws_url).await.unwrap();
// Provoke output. The bash process should echo this back via the PTY.
ws.send(WsMessage::Binary(b"echo hello-broker\n".to_vec()))
ws.send(WsMessage::Binary(b"echo hello-attach\n".to_vec()))
.await
.unwrap();
@@ -375,7 +233,7 @@ async fn broker_ws_attaches_to_local_session() {
loop {
match ws.next().await {
Some(Ok(WsMessage::Binary(bytes))) => {
if String::from_utf8_lossy(&bytes).contains("hello-broker") {
if String::from_utf8_lossy(&bytes).contains("hello-attach") {
return true;
}
}
@@ -386,12 +244,12 @@ async fn broker_ws_attaches_to_local_session() {
})
.await
.unwrap_or(false);
assert!(saw_hello, "PTY did not echo through the broker WS");
assert!(saw_hello, "PTY did not echo through the attach WS");
}
#[tokio::test]
async fn broker_ws_for_unknown_session_returns_404_not_502() {
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
async fn attach_ws_for_unknown_session_returns_404() {
let h = spawn("naps-laptop").await;
let ws_url = format!("ws://{}/api/attach/{}", h.addr, Uuid::new_v4());
let err = connect_async(ws_url)
.await