From de097cf17f840aefba8f0b6cd81512b6298db0bb Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 10:21:27 +0100 Subject: [PATCH] test(maestro): rework e2e harness for a single daemon Drops the registry subprocess and the T2 broker scenarios; the harness spawns one daemon and the client talks straight to its /api/* surface. --- crates/maestro-e2e/src/client.rs | 97 +--- crates/maestro-e2e/src/daemon.rs | 170 ++---- crates/maestro-e2e/src/lib.rs | 10 +- crates/maestro-e2e/tests/e2e.rs | 612 ++------------------ crates/maestro-e2e/tests/tui_integration.rs | 46 +- crates/maestro-tui/src/sidebar.rs | 5 +- 6 files changed, 125 insertions(+), 815 deletions(-) diff --git a/crates/maestro-e2e/src/client.rs b/crates/maestro-e2e/src/client.rs index 99aa0f2..223efe4 100644 --- a/crates/maestro-e2e/src/client.rs +++ b/crates/maestro-e2e/src/client.rs @@ -1,15 +1,13 @@ -//! Thin HTTP+WS client used by tests to drive the registry. -//! -//! Cookies are managed automatically by `reqwest`'s built-in cookie store. +//! Thin HTTP+WS client used by tests to drive a daemon's `/api/*` surface. #![allow(clippy::result_large_err)] +use std::sync::Arc; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; -use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, RegistryEvent, Session}; +use maestro_protocol::{CatalogEvent, CreateSessionRequest, Session}; use reqwest::Client as HttpClient; -use std::sync::Arc; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::Message as WsMessage; use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; @@ -28,7 +26,7 @@ pub enum ClientError { Status { status: u16, body: String }, } -/// Sticky-cookie HTTP client + a few WS-aware helpers. +/// HTTP client plus a few WS-aware helpers, bound to one daemon. #[derive(Debug, Clone)] pub struct TestClient { http_base: String, @@ -39,7 +37,6 @@ pub struct TestClient { impl TestClient { pub fn new(http_base: &str, ws_base: &str) -> Self { let http = HttpClient::builder() - .cookie_store(true) .timeout(Duration::from_secs(5)) .build() .expect("reqwest client"); @@ -50,23 +47,6 @@ impl TestClient { } } - /// Hit any GET endpoint so the registry issues the auth cookie that all - /// subsequent calls reuse. - pub async fn bootstrap(&self) -> Result<(), ClientError> { - let _ = self.list_daemons().await?; - Ok(()) - } - - pub async fn list_daemons(&self) -> Result, ClientError> { - let r = self - .http - .get(format!("{}/api/daemons", self.http_base)) - .send() - .await?; - check_ok(&r)?; - Ok(r.json().await?) - } - pub async fn list_sessions(&self) -> Result, ClientError> { let r = self .http @@ -105,50 +85,23 @@ impl TestClient { Ok(()) } - pub async fn connect_info(&self, id: Uuid) -> Result { - let r = self - .http - .get(format!("{}/api/sessions/{id}/connect-info", self.http_base)) - .send() - .await?; - check_ok(&r)?; - Ok(r.json().await?) - } - /// Connect to the daemon's per-session WS and return the stream. The /// caller drives the conversation. pub async fn attach( &self, - info: &ConnectInfo, id: Uuid, ) -> Result>, ClientError> { - let url = format!( - "{}/sessions/{id}/stream?token={}", - info.daemon_url.trim_end_matches('/'), - info.jwt - ); + let url = format!("{}/api/attach/{id}", self.ws_base); let request = url.into_client_request()?; let (ws, _resp) = connect_async(request).await?; Ok(ws) } - /// Subscribe to the registry's `/api/events` WS. Returns a tokio task - /// handle that drives the WS and pushes parsed events into an Arc>. + /// Subscribe to the daemon's `/api/events` WS. Returns a handle whose + /// background task pushes parsed events into a shared vector. pub async fn subscribe_events(&self) -> EventStream { - // Forward the latest cookie value with the upgrade — `reqwest` and - // `tungstenite` don't share cookie jars, so we copy the cookie out by - // hand. The registry will accept us either way (missing cookie just - // gets a fresh one), but reusing the same identity keeps log noise - // down. - let cookie_header = self.grab_cookie().await.unwrap_or_default(); let url = format!("{}/api/events", self.ws_base); - let mut request = url.into_client_request().expect("ws req"); - if !cookie_header.is_empty() { - request.headers_mut().insert( - reqwest::header::COOKIE, - cookie_header.parse().expect("cookie hv"), - ); - } + let request = url.into_client_request().expect("ws req"); let (ws, _resp) = connect_async(request).await.expect("events ws"); let events = Arc::new(tokio::sync::Mutex::new(Vec::new())); @@ -158,7 +111,7 @@ impl TestClient { while let Some(msg) = stream.next().await { match msg { Ok(WsMessage::Text(t)) => { - if let Ok(ev) = serde_json::from_str::(&t) { + if let Ok(ev) = serde_json::from_str::(&t) { stream_events.lock().await.push(ev); } } @@ -170,34 +123,6 @@ impl TestClient { }); EventStream { events, task } } - - async fn grab_cookie(&self) -> Option { - // Force a request to make sure the jar has a cookie, then read it back - // via a debug header roundtrip. reqwest's CookieStore isn't directly - // exposed; the simplest path is to ask the registry for an auth - // cookie and capture the Set-Cookie from the response. - let r = self - .http - .get(format!("{}/api/daemons", self.http_base)) - .send() - .await - .ok()?; - // The cookie store has the cookie now, but reqwest doesn't surface - // it. We rebuild from Set-Cookie if present. - r.headers() - .get_all(reqwest::header::SET_COOKIE) - .iter() - .find_map(|hv| hv.to_str().ok().and_then(extract_cookie_pair)) - } -} - -fn extract_cookie_pair(set_cookie: &str) -> Option { - // Take everything before the first ';', which is `name=value`. - set_cookie - .split(';') - .next() - .map(str::trim) - .map(String::from) } fn check_ok(r: &reqwest::Response) -> Result<(), ClientError> { @@ -214,12 +139,12 @@ fn check_ok(r: &reqwest::Response) -> Result<(), ClientError> { /// Handle to a background task draining the events WS into a vector. #[derive(Debug)] pub struct EventStream { - events: Arc>>, + events: Arc>>, task: tokio::task::JoinHandle<()>, } impl EventStream { - pub async fn snapshot(&self) -> Vec { + pub async fn snapshot(&self) -> Vec { self.events.lock().await.clone() } } diff --git a/crates/maestro-e2e/src/daemon.rs b/crates/maestro-e2e/src/daemon.rs index d48a23b..7dac460 100644 --- a/crates/maestro-e2e/src/daemon.rs +++ b/crates/maestro-e2e/src/daemon.rs @@ -1,4 +1,4 @@ -//! Subprocess wrapper around `maestrod`. +//! Subprocess wrapper around `maestro daemon`. use std::net::{Ipv4Addr, SocketAddr}; use std::path::PathBuf; @@ -10,68 +10,49 @@ use tempfile::TempDir; use crate::build::workspace_target_bin; use crate::util::{free_port, wait_for_async}; -/// A live `maestrod` running in a child process. Kill the child on drop. +/// A live daemon running in a child process. Kill the child on drop. /// -/// `state_dir`, `home`, and `tmux_tmpdir` are kept as `TempDir`s so they -/// outlive the process and get cleaned up automatically. +/// `home` and `tmux_tmpdir` are kept as `TempDir`s so they outlive the process +/// and get cleaned up automatically. State lives under `home` via +/// `XDG_STATE_HOME`. #[derive(Debug)] pub struct Daemon { - pub daemon_id: String, pub bind_addr: SocketAddr, - pub daemon_url: String, child: Option, home: TempDir, - state: TempDir, tmux: TempDir, - registry_url: String, - token: String, -} - -/// Optional knobs for `Daemon::spawn_with` — perimeter advertisement for -/// NAP-247 broker tests. -#[derive(Debug, Default, Clone)] -pub struct DaemonOptions { - pub perimeter_id: Option, - pub private_addr: Option, } impl Daemon { - /// Spawn a fresh daemon pointing at `registry_base_url` (the registry's - /// HTTP base, e.g. `http://127.0.0.1:1234`). The daemon derives its WS - /// endpoint internally via `derive_daemon_ws_url`. - pub async fn spawn(daemon_id: &str, registry_base_url: &str, token: &str) -> Self { - Self::spawn_with( - daemon_id, - registry_base_url, - token, - DaemonOptions::default(), - ) - .await + /// Spawn a fresh daemon on a free localhost port. + pub async fn spawn() -> Self { + Self::spawn_at(SocketAddr::from((Ipv4Addr::LOCALHOST, free_port()))).await } - /// Spawn with NAP-247 options (perimeter advertisement). - pub async fn spawn_with( - daemon_id: &str, - registry_base_url: &str, - token: &str, - opts: DaemonOptions, - ) -> Self { - let port = free_port(); - let bind = SocketAddr::from((Ipv4Addr::LOCALHOST, port)); - Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await + /// Spawn on a caller-supplied bind address. + pub async fn spawn_at(bind: SocketAddr) -> Self { + let home = TempDir::new().expect("home tmpdir"); + let tmux = TempDir::new().expect("tmux tmpdir"); + + let child = launch(bind, home.path(), tmux.path()); + let me = Self { + bind_addr: bind, + child: Some(child), + home, + tmux, + }; + me.wait_until_ready().await; + me } - /// Spawn on a caller-supplied bind address. Used by NAP-247 broker tests - /// where the daemon's advertised `private_addr` must equal the actual - /// bind address so the broker can dial it back. - pub async fn spawn_at( - daemon_id: &str, - registry_base_url: &str, - token: &str, - bind: SocketAddr, - opts: DaemonOptions, - ) -> Self { - Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await + /// HTTP base URL of this daemon's API. + pub fn http_base(&self) -> String { + format!("http://{}", self.bind_addr) + } + + /// WS base URL of this daemon's API. + pub fn ws_base(&self) -> String { + format!("ws://{}", self.bind_addr) } /// Stop the underlying process, but keep its state dirs around so we can @@ -85,70 +66,18 @@ impl Daemon { /// Re-spawn the daemon on the same port + state. The previous tmux /// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt - /// its sessions. Restarts inherit the original options via the on-disk - /// config file written at `spawn_with_port_and_opts` time. + /// its sessions. pub async fn restart(&mut self) { self.stop(); - let child = launch( - &self.daemon_id, - self.bind_addr, - &self.daemon_url, - &self.registry_url, - &self.token, - self.home.path(), - self.state.path(), - self.tmux.path(), - &DaemonOptions::default(), - ); - self.child = Some(child); - self.wait_until_attach_ready().await; + self.child = Some(launch(self.bind_addr, self.home.path(), self.tmux.path())); + self.wait_until_ready().await; } - async fn spawn_with_port_and_opts( - daemon_id: &str, - registry_url: &str, - token: &str, - bind: SocketAddr, - opts: DaemonOptions, - ) -> Self { - let home = TempDir::new().expect("home tmpdir"); - let state = TempDir::new().expect("state tmpdir"); - let tmux = TempDir::new().expect("tmux tmpdir"); - - let daemon_url = format!("ws://{bind}"); - let child = launch( - daemon_id, - bind, - &daemon_url, - registry_url, - token, - home.path(), - state.path(), - tmux.path(), - &opts, - ); - let me = Self { - daemon_id: daemon_id.to_string(), - bind_addr: bind, - daemon_url, - child: Some(child), - home, - state, - tmux, - registry_url: registry_url.to_string(), - token: token.to_string(), - }; - me.wait_until_attach_ready().await; - me - } - - async fn wait_until_attach_ready(&self) { + async fn wait_until_ready(&self) { let addr = self.bind_addr; - wait_for_async( - "daemon attach port", - Duration::from_secs(5), - || async move { tokio::net::TcpStream::connect(addr).await.is_ok() }, - ) + wait_for_async("daemon api port", Duration::from_secs(5), || async move { + tokio::net::TcpStream::connect(addr).await.is_ok() + }) .await; } @@ -158,7 +87,7 @@ impl Daemon { pub fn state_dir(&self) -> PathBuf { // Matches maestro_daemon::default_state_dir which uses XDG_STATE_HOME. - self.home.path().join(".local/state/maestrod") + self.home.path().join(".local/state/maestro") } } @@ -171,18 +100,7 @@ impl Drop for Daemon { } } -#[allow(clippy::too_many_arguments)] -fn launch( - daemon_id: &str, - bind: SocketAddr, - daemon_url: &str, - registry_url: &str, - token: &str, - home: &std::path::Path, - _state: &std::path::Path, - tmux_tmpdir: &std::path::Path, - opts: &DaemonOptions, -) -> Child { +fn launch(bind: SocketAddr, home: &std::path::Path, tmux_tmpdir: &std::path::Path) -> Child { let bin = workspace_target_bin("maestro", "maestro"); let mut cmd = Command::new(&bin); cmd.arg("daemon"); @@ -195,21 +113,11 @@ fn launch( .env("XDG_CONFIG_HOME", home.join(".config")) .env("XDG_STATE_HOME", home.join(".local/state")) .env("TMUX_TMPDIR", tmux_tmpdir) - .env("MAESTRO_TOKEN", token) - .env("MAESTRO_DAEMON_ID", daemon_id) - .env("MAESTRO_REGISTRY_URL", registry_url) .env("MAESTRO_DAEMON_BIND_ADDR", bind.to_string()) - .env("MAESTRO_DAEMON_URL", daemon_url) .env("RUST_LOG", "warn") .stdout(Stdio::null()) .stderr(Stdio::inherit()) .arg("run"); - if let Some(p) = &opts.perimeter_id { - cmd.env("MAESTRO_DAEMON_PERIMETER_ID", p); - } - if let Some(a) = &opts.private_addr { - cmd.env("MAESTRO_DAEMON_PRIVATE_ADDR", a); - } cmd.spawn() .unwrap_or_else(|e| panic!("spawn {} failed: {e}", bin.display())) } diff --git a/crates/maestro-e2e/src/lib.rs b/crates/maestro-e2e/src/lib.rs index eed3af1..9096913 100644 --- a/crates/maestro-e2e/src/lib.rs +++ b/crates/maestro-e2e/src/lib.rs @@ -1,8 +1,8 @@ //! Cross-process integration test harness for Maestro. //! -//! Each test spawns a real `maestro-registry` and one or more `maestrod` -//! subprocesses, isolates them in temp dirs (`$HOME`, `TMUX_TMPDIR`, state -//! dir), and drives them through their public HTTP/WS APIs. +//! Each test spawns a real `maestro daemon` subprocess, isolates it in temp +//! dirs (`$HOME`, `TMUX_TMPDIR`, state dir), and drives it through its public +//! HTTP/WS API. //! //! See `tests/` for the actual scenarios. @@ -11,10 +11,8 @@ pub mod build; pub mod client; pub mod daemon; -pub mod registry; pub mod util; pub use build::workspace_target_bin; pub use client::TestClient; -pub use daemon::{Daemon, DaemonOptions}; -pub use registry::Registry; +pub use daemon::Daemon; diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 5de053e..609560d 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -1,9 +1,8 @@ //! Cross-process end-to-end tests for the headless Maestro slice. //! -//! Each test brings up a real `maestro-registry` plus 1-2 real `maestrod` -//! subprocesses, isolates them in temp dirs, and drives them through the -//! public HTTP+WS surface using the same client APIs a real CLI/web client -//! would. +//! Each test brings up a real daemon subprocess, isolated in temp dirs, and +//! drives it through the public HTTP+WS surface using the same client APIs a +//! real CLI/web client would. //! //! These tests require `tmux` and `git` on `$PATH`. @@ -15,45 +14,19 @@ use std::time::{Duration, Instant}; use futures_util::{SinkExt, StreamExt}; use maestro_e2e::util::wait_for_async; -use maestro_e2e::{Daemon, DaemonOptions, Registry, TestClient}; +use maestro_e2e::{Daemon, TestClient}; use maestro_protocol::{CreateSessionRequest, WorktreeSpec}; use tempfile::TempDir; use tokio_tungstenite::tungstenite::Message as WsMessage; -const TOKEN: &str = "e2e-shared-token"; - -async fn bring_up(daemon_ids: &[&str]) -> (Registry, Vec, TestClient) { - let registry = Registry::spawn(TOKEN).await; - let mut daemons = Vec::with_capacity(daemon_ids.len()); - for id in daemon_ids { - let d = Daemon::spawn(id, ®istry.http_base(), TOKEN).await; - daemons.push(d); - } - let client = TestClient::new(®istry.http_base(), ®istry.ws_base()); - client.bootstrap().await.expect("bootstrap"); - - // Wait until both daemons have connected and announced themselves. - let expected: Vec = daemon_ids.iter().map(|s| s.to_string()).collect(); - wait_for_async("daemons to register", Duration::from_secs(10), || { - let client = client.clone(); - let expected = expected.clone(); - async move { - match client.list_daemons().await { - Ok(ds) => expected - .iter() - .all(|id| ds.iter().any(|d| &d.daemon_id == id)), - Err(_) => false, - } - } - }) - .await; - - (registry, daemons, client) +async fn bring_up() -> (Daemon, TestClient) { + let daemon = Daemon::spawn().await; + let client = TestClient::new(&daemon.http_base(), &daemon.ws_base()); + (daemon, client) } -fn create_req(daemon_id: &str, cwd: &std::path::Path, command: &str) -> CreateSessionRequest { +fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest { CreateSessionRequest { - daemon_id: daemon_id.into(), cwd: cwd.to_path_buf(), command: command.into(), env: HashMap::new(), @@ -63,211 +36,39 @@ fn create_req(daemon_id: &str, cwd: &std::path::Path, command: &str) -> CreateSe } #[tokio::test(flavor = "multi_thread")] -async fn lists_sessions_from_both_daemons() { - let (_reg, _daemons, client) = bring_up(&["alpha", "beta"]).await; +async fn lists_spawned_sessions() { + let (_daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let a = client - .spawn_session(&create_req("alpha", cwd.path(), "sleep 30")) + .spawn_session(&create_req(cwd.path(), "sleep 30")) .await - .expect("spawn alpha"); + .expect("spawn a"); let b = client - .spawn_session(&create_req("beta", cwd.path(), "sleep 30")) + .spawn_session(&create_req(cwd.path(), "sleep 30")) .await - .expect("spawn beta"); + .expect("spawn b"); let sessions = client.list_sessions().await.expect("list"); - assert!(sessions - .iter() - .any(|s| s.id == a.id && s.daemon_id == "alpha")); - assert!(sessions - .iter() - .any(|s| s.id == b.id && s.daemon_id == "beta")); + assert!(sessions.iter().any(|s| s.id == a.id)); + assert!(sessions.iter().any(|s| s.id == b.id)); client.kill_session(a.id).await.expect("kill a"); client.kill_session(b.id).await.expect("kill b"); } -#[tokio::test(flavor = "multi_thread")] -async fn daemon_local_api_mirrors_peer_catalog() { - // NAP-246: a daemon's GET /api/daemons + /api/sessions should reflect - // peer state (not just self), because the registry fans out catalog - // events to every connected daemon. - let (_reg, daemons, client) = bring_up(&["alpha", "beta"]).await; - let cwd = TempDir::new().unwrap(); - - // Create a session on beta via the registry path. - let b = client - .spawn_session(&create_req("beta", cwd.path(), "sleep 30")) - .await - .expect("spawn beta"); - - // Now query alpha's local HTTP API and confirm it sees beta + beta's - // session. The mirror is eventually consistent, so poll. - let alpha_addr = daemons[0].bind_addr; - let alpha_base = format!("http://{alpha_addr}"); - - wait_for_async( - "alpha mirrors beta in /api/daemons", - Duration::from_secs(5), - || { - let base = alpha_base.clone(); - async move { - let resp = match reqwest::get(format!("{base}/api/daemons")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|d| d.daemon_id == "beta") - } - }, - ) - .await; - - wait_for_async( - "alpha mirrors beta's session in /api/sessions", - Duration::from_secs(5), - || { - let base = alpha_base.clone(); - let want = b.id; - async move { - let resp = match reqwest::get(format!("{base}/api/sessions")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|s| s.id == want && s.daemon_id == "beta") - } - }, - ) - .await; - - // Both self and peer come back. - let list: Vec = reqwest::get(format!("{alpha_base}/api/daemons")) - .await - .unwrap() - .json() - .await - .unwrap(); - let ids: Vec<&str> = list.iter().map(|d| d.daemon_id.as_str()).collect(); - assert!(ids.contains(&"alpha")); - assert!(ids.contains(&"beta")); - - // NAP-328: home_dir is advertised end-to-end — for a peer relayed through - // the registry (beta, via Hello → register → catalog mirror), so the web - // spawn form can prefill the working directory. - let beta = list.iter().find(|d| d.daemon_id == "beta").unwrap(); - assert!( - !beta.home_dir.is_empty(), - "peer home_dir should propagate through the registry: {beta:?}" - ); - - client.kill_session(b.id).await.expect("kill b"); -} - -#[tokio::test(flavor = "multi_thread")] -async fn home_daemon_brokers_cross_daemon_create_and_kill() { - // NAP-327: POST /api/sessions to alpha (home) with daemon_id=beta must - // broker the create to beta over the registry control link and return - // beta's session — and a kill via alpha must reach beta too. This is the - // path maestro-web uses (it only talks to its local daemon). - let (_reg, daemons, registry_client) = bring_up(&["alpha", "beta"]).await; - let cwd = TempDir::new().unwrap(); - - // A client pointed at alpha's *local* API (not the registry). - let alpha_base = format!("http://{}", daemons[0].bind_addr); - let alpha = TestClient::new(&alpha_base, &alpha_base); - - // Create a beta-owned session by asking alpha. - let s = alpha - .spawn_session(&create_req("beta", cwd.path(), "sleep 30")) - .await - .expect("alpha brokers create to beta"); - assert_eq!(s.daemon_id, "beta", "session should be owned by beta"); - - // The registry catalog (and thus every client) sees it on beta. - wait_for_async( - "beta session visible in catalog", - Duration::from_secs(5), - || { - let client = registry_client.clone(); - let want = s.id; - async move { - client - .list_sessions() - .await - .map(|ss| ss.iter().any(|x| x.id == want && x.daemon_id == "beta")) - .unwrap_or(false) - } - }, - ) - .await; - - // alpha's mirror must reflect beta's session before it can broker the kill. - wait_for_async( - "alpha mirrors beta's session", - Duration::from_secs(5), - || { - let base = alpha_base.clone(); - let want = s.id; - async move { - let resp = match reqwest::get(format!("{base}/api/sessions")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|x| x.id == want) - } - }, - ) - .await; - - // Kill the beta session via alpha — brokered to beta. - alpha - .kill_session(s.id) - .await - .expect("alpha brokers kill to beta"); - - wait_for_async( - "beta session gone from catalog", - Duration::from_secs(5), - || { - let client = registry_client.clone(); - let gone = s.id; - async move { - client - .list_sessions() - .await - .map(|ss| !ss.iter().any(|x| x.id == gone)) - .unwrap_or(false) - } - }, - ) - .await; -} - #[tokio::test(flavor = "multi_thread")] async fn multi_attach_sees_same_output() { - let (_reg, _daemons, client) = bring_up(&["solo"]).await; + let (_daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let s = client - .spawn_session(&create_req("solo", cwd.path(), "bash --noprofile --norc")) + .spawn_session(&create_req(cwd.path(), "bash --noprofile --norc")) .await .expect("spawn"); - let info = client.connect_info(s.id).await.expect("connect-info"); - let mut a = client.attach(&info, s.id).await.expect("attach a"); - let mut b = client.attach(&info, s.id).await.expect("attach b"); + let mut a = client.attach(s.id).await.expect("attach a"); + let mut b = client.attach(s.id).await.expect("attach b"); // Drive input through client A; both attachments should observe it via // the daemon's PTY broadcast. @@ -314,13 +115,12 @@ async fn multi_attach_sees_same_output() { #[tokio::test(flavor = "multi_thread")] async fn cleanup_on_kill_removes_worktree() { - let (_reg, _daemons, client) = bring_up(&["host"]).await; + let (_daemon, client) = bring_up().await; let base = TempDir::new().unwrap(); init_repo(base.path()); let req = CreateSessionRequest { - daemon_id: "host".into(), cwd: base.path().to_path_buf(), command: "sleep 30".into(), env: HashMap::new(), @@ -362,52 +162,20 @@ async fn cleanup_on_kill_removes_worktree() { #[tokio::test(flavor = "multi_thread")] async fn daemon_restart_recovers_session() { - let (_reg, mut daemons, client) = bring_up(&["restarter"]).await; + let (mut daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let s = client - .spawn_session(&create_req("restarter", cwd.path(), "sleep 300")) + .spawn_session(&create_req(cwd.path(), "sleep 300")) .await .expect("spawn"); - assert_eq!(s.daemon_id, "restarter"); - // Stop the daemon and wait for the registry to notice the disconnect. - // The catalog drops sessions for a daemon that's no longer connected. - daemons[0].stop(); - wait_for_async( - "registry drops disconnected daemon", - Duration::from_secs(10), - || { - let client = client.clone(); - async move { - client - .list_daemons() - .await - .map(|ds| ds.iter().all(|d| d.daemon_id != "restarter")) - .unwrap_or(false) - } - }, - ) - .await; - - // Bring the daemon back. It should recover its tmux session and the - // registry catalog should repopulate it. - daemons[0].restart().await; - - wait_for_async("restarter re-registers", Duration::from_secs(10), || { - let client = client.clone(); - async move { - client - .list_daemons() - .await - .map(|ds| ds.iter().any(|d| d.daemon_id == "restarter")) - .unwrap_or(false) - } - }) - .await; + // Restart against the same tmux server + state dir: the daemon must adopt + // the still-running tmux session on boot. + daemon.restart().await; wait_for_async( - "session reappears in catalog", + "session reappears after restart", Duration::from_secs(10), || { let client = client.clone(); @@ -428,7 +196,7 @@ async fn daemon_restart_recovers_session() { #[tokio::test(flavor = "multi_thread")] async fn concurrent_spawns_each_get_unique_ids() { - let (_reg, _daemons, client) = bring_up(&["parallel"]).await; + let (_daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let mut handles = Vec::new(); @@ -437,7 +205,7 @@ async fn concurrent_spawns_each_get_unique_ids() { let path = cwd.path().to_path_buf(); handles.push(tokio::spawn(async move { client - .spawn_session(&create_req("parallel", &path, "sleep 30")) + .spawn_session(&create_req(&path, "sleep 30")) .await .expect("spawn") })); @@ -461,8 +229,8 @@ async fn spawn_records_history_and_infers_gitroot_for_form() { // the daemon's recent-dirs endpoint surfaces on the spawn form's next open // (#27's recents picker), and gitroot infers the worktree base_repo from the // same dir. Exercised end to end against real tmux + git. - let (_reg, daemons, client) = bring_up(&["host"]).await; - let daemon_base = format!("http://{}", daemons[0].bind_addr); + let (daemon, client) = bring_up().await; + let daemon_base = daemon.http_base(); // A git repo as the working dir, so gitroot has a repo to infer. let repo = TempDir::new().unwrap(); @@ -471,7 +239,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() { // Spawn in that dir (the command stands in for a preset tile selection). let s = client - .spawn_session(&create_req("host", repo.path(), "sleep 30")) + .spawn_session(&create_req(repo.path(), "sleep 30")) .await .expect("spawn"); @@ -482,7 +250,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() { "recent-dirs surfaces the spawned cwd", Duration::from_secs(5), || { - let url = format!("{daemon_base}/api/daemons/host/recent-dirs"); + let url = format!("{daemon_base}/api/recent-dirs"); let want = cwd.clone(); async move { let resp = match reqwest::get(&url).await { @@ -505,7 +273,7 @@ async fn spawn_records_history_and_infers_gitroot_for_form() { // gitroot infers the worktree base_repo from the same dir (#25/#26): the dir // is itself the repo toplevel (it holds `.git`). let gr: maestro_protocol::GitRoot = reqwest::Client::new() - .get(format!("{daemon_base}/api/daemons/host/gitroot")) + .get(format!("{daemon_base}/api/gitroot")) .query(&[("path", cwd.as_str())]) .send() .await @@ -530,15 +298,15 @@ async fn activity_reflects_foreground_and_input_blocking() { // (the distinction the /proc//syscall check buys us). Both report the // right foreground. Exercised against real tmux. use maestro_protocol::ActivityState; - let (_reg, _daemons, client) = bring_up(&["host"]).await; + let (_daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let cat = client - .spawn_session(&create_req("host", cwd.path(), "cat")) + .spawn_session(&create_req(cwd.path(), "cat")) .await .expect("spawn cat"); let slp = client - .spawn_session(&create_req("host", cwd.path(), "sleep 300")) + .spawn_session(&create_req(cwd.path(), "sleep 300")) .await .expect("spawn sleep"); @@ -621,244 +389,14 @@ fn git_worktree_list(dir: &std::path::Path) -> Vec { .collect() } -#[tokio::test(flavor = "multi_thread")] -async fn broker_t2_attach_through_home_daemon() { - // NAP-247: two daemons share perimeter "wg-test". alpha's broker resolves - // a session owned by beta, gets a pair JWT + perimeter address from the - // registry, dials beta via T2, and proxies the attach. - let registry = Registry::spawn("e2e-shared-token").await; - let beta_bind = std::net::SocketAddr::from(( - std::net::Ipv4Addr::LOCALHOST, - maestro_e2e::util::free_port(), - )); - let beta = Daemon::spawn_at( - "beta", - ®istry.http_base(), - "e2e-shared-token", - beta_bind, - DaemonOptions { - perimeter_id: Some("wg-test".into()), - private_addr: Some(beta_bind.to_string()), - }, - ) - .await; - let alpha_bind = std::net::SocketAddr::from(( - std::net::Ipv4Addr::LOCALHOST, - maestro_e2e::util::free_port(), - )); - let alpha = Daemon::spawn_at( - "alpha", - ®istry.http_base(), - "e2e-shared-token", - alpha_bind, - DaemonOptions { - perimeter_id: Some("wg-test".into()), - private_addr: Some(alpha_bind.to_string()), - }, - ) - .await; - assert_eq!(beta.bind_addr, beta_bind); - assert_eq!(alpha.bind_addr, alpha_bind); - - let client = TestClient::new(®istry.http_base(), ®istry.ws_base()); - client.bootstrap().await.expect("bootstrap"); - wait_for_async("both daemons connected", Duration::from_secs(10), || { - let client = client.clone(); - async move { - client - .list_daemons() - .await - .map(|ds| { - ds.iter().any(|d| d.daemon_id == "alpha") - && ds.iter().any(|d| d.daemon_id == "beta") - }) - .unwrap_or(false) - } - }) - .await; - - let cwd = TempDir::new().unwrap(); - let session = client - .spawn_session(&create_req("beta", cwd.path(), "bash --noprofile --norc")) - .await - .expect("spawn beta"); - - // Wait until alpha's catalog mirror sees the peer session — otherwise the - // broker can't locate the owning daemon. - let alpha_base = format!("http://{}", alpha.bind_addr); - let sid = session.id; - wait_for_async( - "alpha mirrors beta's session", - Duration::from_secs(5), - || { - let base = alpha_base.clone(); - async move { - let resp = match reqwest::get(format!("{base}/api/sessions")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|s| s.id == sid && s.daemon_id == "beta") - } - }, - ) - .await; - - // Attach through alpha's broker. - let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid); - let (mut ws, _) = tokio_tungstenite::connect_async(broker_url) - .await - .expect("broker WS handshake"); - ws.send(WsMessage::Binary(b"echo MAESTRO_BROKER_E2E\n".to_vec())) - .await - .expect("send keys"); - - let want = b"MAESTRO_BROKER_E2E"; - let mut buf = Vec::new(); - let deadline = std::time::Instant::now() + Duration::from_secs(5); - while std::time::Instant::now() < deadline { - tokio::select! { - msg = ws.next() => match msg { - Some(Ok(WsMessage::Binary(b))) => buf.extend(b), - Some(Ok(_)) => {} - _ => break, - }, - _ = tokio::time::sleep(Duration::from_millis(50)) => {} - } - if buf.windows(want.len()).any(|w| w == want) { - break; - } - } - assert!( - buf.windows(want.len()).any(|w| w == want), - "broker WS never echoed our keystrokes: buf len {}", - buf.len() - ); - - let _ = ws.close(None).await; - client.kill_session(sid).await.expect("kill"); - drop(alpha); - drop(beta); - drop(registry); -} - -#[tokio::test(flavor = "multi_thread")] -async fn broker_attach_returns_502_on_perimeter_mismatch() { - // NAP-270: the asymmetric-VLAN boundary. alpha and beta are on DIFFERENT - // perimeters, so alpha's broker must refuse T2 (the only implemented - // transport) with a structured 502 rather than dial across perimeters. - let registry = Registry::spawn("e2e-shared-token").await; - let beta_bind = std::net::SocketAddr::from(( - std::net::Ipv4Addr::LOCALHOST, - maestro_e2e::util::free_port(), - )); - let beta = Daemon::spawn_at( - "beta", - ®istry.http_base(), - "e2e-shared-token", - beta_bind, - DaemonOptions { - perimeter_id: Some("wg-beta".into()), - private_addr: Some(beta_bind.to_string()), - }, - ) - .await; - let alpha_bind = std::net::SocketAddr::from(( - std::net::Ipv4Addr::LOCALHOST, - maestro_e2e::util::free_port(), - )); - let alpha = Daemon::spawn_at( - "alpha", - ®istry.http_base(), - "e2e-shared-token", - alpha_bind, - DaemonOptions { - // Different perimeter from beta → T2 unavailable. - perimeter_id: Some("wg-alpha".into()), - private_addr: Some(alpha_bind.to_string()), - }, - ) - .await; - - let client = TestClient::new(®istry.http_base(), ®istry.ws_base()); - client.bootstrap().await.expect("bootstrap"); - wait_for_async("both daemons connected", Duration::from_secs(10), || { - let client = client.clone(); - async move { - client - .list_daemons() - .await - .map(|ds| ds.len() >= 2) - .unwrap_or(false) - } - }) - .await; - - let cwd = TempDir::new().unwrap(); - let session = client - .spawn_session(&create_req("beta", cwd.path(), "sleep 30")) - .await - .expect("spawn beta"); - let sid = session.id; - - // Wait until alpha's catalog mirror sees beta's session. - let alpha_base = format!("http://{}", alpha.bind_addr); - wait_for_async( - "alpha mirrors beta's session", - Duration::from_secs(5), - || { - let base = alpha_base.clone(); - async move { - let resp = match reqwest::get(format!("{base}/api/sessions")).await { - Ok(r) => r, - Err(_) => return false, - }; - let list: Vec = match resp.json().await { - Ok(v) => v, - Err(_) => return false, - }; - list.iter().any(|s| s.id == sid) - } - }, - ) - .await; - - // Attach through alpha's broker — must 502 (perimeter mismatch), never - // upgrade and never dial beta. - let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid); - let err = tokio_tungstenite::connect_async(broker_url) - .await - .expect_err("cross-perimeter attach must fail the handshake"); - match err { - tokio_tungstenite::tungstenite::Error::Http(resp) => { - assert_eq!( - resp.status(), - 502, - "expected BAD_GATEWAY for perimeter mismatch, got {}", - resp.status() - ); - } - other => panic!("expected an HTTP 502 response, got {other:?}"), - } - - client.kill_session(sid).await.expect("kill"); - drop(alpha); - drop(beta); - drop(registry); -} - #[tokio::test(flavor = "multi_thread")] async fn daemon_events_ws_streams_snapshot_then_session_added() { - // NAP-248: subscribe to a daemon's WS /api/events, expect an initial + // NAP-248: subscribe to the daemon's WS /api/events, expect an initial // Snapshot, then a SessionAdded after we spawn a session via REST. - let (_reg, daemons, client) = bring_up(&["alpha"]).await; + let (daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); - let alpha_addr = daemons[0].bind_addr; - let ws_url = format!("ws://{alpha_addr}/api/events"); + let ws_url = format!("{}/api/events", daemon.ws_base()); let (mut ws, _) = tokio_tungstenite::connect_async(ws_url) .await .expect("events WS handshake"); @@ -866,23 +404,22 @@ async fn daemon_events_ws_streams_snapshot_then_session_added() { // First frame must be a Snapshot (the WS sends it eagerly on connect). let snapshot = next_event(&mut ws).await; match &snapshot { - maestro_protocol::RegistryEvent::Snapshot { daemons, sessions } => { - assert!(daemons.iter().any(|d| d.daemon_id == "alpha")); + maestro_protocol::CatalogEvent::Snapshot { sessions } => { assert!(sessions.is_empty()); } other => panic!("expected initial Snapshot, got {other:?}"), } let session = client - .spawn_session(&create_req("alpha", cwd.path(), "sleep 30")) + .spawn_session(&create_req(cwd.path(), "sleep 30")) .await - .expect("spawn alpha"); + .expect("spawn"); // SessionAdded must show up on the WS within a reasonable window. let saw_added = tokio::time::timeout(Duration::from_secs(5), async { loop { match next_event(&mut ws).await { - maestro_protocol::RegistryEvent::SessionAdded { session: s } + maestro_protocol::CatalogEvent::SessionAdded { session: s } if s.id == session.id => { return true; @@ -902,7 +439,7 @@ async fn daemon_events_ws_streams_snapshot_then_session_added() { } // --------------------------------------------------------------------------- -// NAP-244 — CLI rewired through the home-daemon broker, with registry fallback +// CLI against the local daemon // --------------------------------------------------------------------------- /// Run the real `maestro` binary with an isolated env and capture its output. @@ -921,18 +458,15 @@ fn run_cli(args: &[&str], envs: &[(&str, &str)]) -> std::process::Output { } #[tokio::test(flavor = "multi_thread")] -async fn cli_list_reads_from_home_daemon() { - // Broker path: point the CLI at the daemon's local API and DON'T configure - // a registry. Success proves the catalog read came from the home daemon — - // the registry path would have errored with NoRegistryUrl. - let (_reg, daemons, client) = bring_up(&["alpha"]).await; +async fn cli_list_reads_from_daemon() { + let (daemon, client) = bring_up().await; let cwd = TempDir::new().unwrap(); let session = client - .spawn_session(&create_req("alpha", cwd.path(), "sleep 30")) + .spawn_session(&create_req(cwd.path(), "sleep 30")) .await .expect("spawn"); - let daemon_url = format!("http://{}", daemons[0].bind_addr); + let daemon_url = daemon.http_base(); let home = TempDir::new().unwrap(); let cfg_path = home.path().join("config.toml"); let out = run_cli( @@ -940,7 +474,7 @@ async fn cli_list_reads_from_home_daemon() { &[ ("HOME", home.path().to_str().unwrap()), ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()), - ("MAESTRO_HOME_DAEMON_URL", &daemon_url), + ("MAESTRO_DAEMON_URL", &daemon_url), ], ); let stdout = String::from_utf8_lossy(&out.stdout); @@ -948,65 +482,35 @@ async fn cli_list_reads_from_home_daemon() { assert!(out.status.success(), "list failed: {stderr}\n{stdout}"); assert!( stdout.contains(&session.id.to_string()), - "session not listed via home daemon: {stdout}" - ); - assert!( - !stderr.contains("falling back"), - "unexpected registry fallback: {stderr}" + "session not listed: {stdout}" ); client.kill_session(session.id).await.ok(); } #[tokio::test(flavor = "multi_thread")] -async fn cli_list_falls_back_to_registry_when_daemon_down() { - // Compat path: home daemon URL points at a dead port (connection refused). - // The CLI must warn and fall back to the configured registry, still - // returning the session. - let (reg, _daemons, client) = bring_up(&["alpha"]).await; - let cwd = TempDir::new().unwrap(); - let session = client - .spawn_session(&create_req("alpha", cwd.path(), "sleep 30")) - .await - .expect("spawn"); - +async fn cli_list_fails_when_daemon_is_down() { let dead_daemon = format!("http://127.0.0.1:{}", maestro_e2e::util::free_port()); - let registry_url = reg.http_base(); let home = TempDir::new().unwrap(); let cfg_path = home.path().join("config.toml"); let out = run_cli( - &["list", "--json"], + &["list"], &[ ("HOME", home.path().to_str().unwrap()), ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()), - ("MAESTRO_HOME_DAEMON_URL", &dead_daemon), - ("MAESTRO_REGISTRY_URL", ®istry_url), + ("MAESTRO_DAEMON_URL", &dead_daemon), ], ); - let stdout = String::from_utf8_lossy(&out.stdout); - let stderr = String::from_utf8_lossy(&out.stderr); - assert!(out.status.success(), "list failed: {stderr}\n{stdout}"); - assert!( - stdout.contains(&session.id.to_string()), - "session not listed via registry fallback: {stdout}" - ); - assert!( - stderr.contains("falling back"), - "expected a fallback warning on stderr: {stderr}" - ); - - client.kill_session(session.id).await.ok(); + assert!(!out.status.success(), "expected a failure with no daemon"); } -async fn next_event(ws: &mut S) -> maestro_protocol::RegistryEvent +async fn next_event(ws: &mut S) -> maestro_protocol::CatalogEvent where S: StreamExt> + Unpin, { loop { match ws.next().await { - Some(Ok(WsMessage::Text(t))) => { - return serde_json::from_str(&t).expect("RegistryEvent") - } + Some(Ok(WsMessage::Text(t))) => return serde_json::from_str(&t).expect("CatalogEvent"), Some(Ok(_)) => continue, other => panic!("WS closed waiting for event: {other:?}"), } diff --git a/crates/maestro-e2e/tests/tui_integration.rs b/crates/maestro-e2e/tests/tui_integration.rs index 4f96c95..540e1f9 100644 --- a/crates/maestro-e2e/tests/tui_integration.rs +++ b/crates/maestro-e2e/tests/tui_integration.rs @@ -1,11 +1,11 @@ //! TUI integration tests (NAP-282). //! //! Exercises `CatalogStore` and the per-session `SessionTask` against a real -//! registry + daemon subprocess pair. We don't spin the full ratatui frontend +//! daemon subprocess. We don't spin the full ratatui frontend //! here — those interactions are best tested with `TestBackend` unit tests in //! `crates/maestro-tui/src/*.rs`. This file fills the gap between those //! and the headless e2e suite by verifying that the TUI's wire layer plays -//! nicely with a real `maestrod`. +//! nicely with a real daemon. //! //! Requires `tmux` and `git` on `$PATH`. @@ -14,26 +14,10 @@ use std::path::PathBuf; use std::time::{Duration, Instant}; -use maestro_e2e::{Daemon, Registry}; +use maestro_e2e::Daemon; use maestro_protocol::CreateSessionRequest; use maestro_tui::catalog::CatalogStore; -const TOKEN: &str = "tui-integration-token"; - -async fn bring_up() -> (Registry, Daemon) { - let registry = Registry::spawn(TOKEN).await; - let daemon = Daemon::spawn("konishi", ®istry.http_base(), TOKEN).await; - (registry, daemon) -} - -fn http_base(daemon: &Daemon) -> String { - format!("http://{}", daemon.bind_addr) -} - -fn ws_base(daemon: &Daemon) -> String { - format!("ws://{}", daemon.bind_addr) -} - async fn wait_until(label: &str, deadline: Duration, mut check: F) where F: FnMut() -> Fut, @@ -53,8 +37,8 @@ where #[tokio::test(flavor = "multi_thread")] async fn catalog_bootstrap_against_real_daemon() { - let (_registry, daemon) = bring_up().await; - let ws_url = format!("{}/api/events", ws_base(&daemon)); + let daemon = Daemon::spawn().await; + let ws_url = format!("{}/api/events", daemon.ws_base()); let store = CatalogStore::default(); let bg = store.clone(); @@ -64,17 +48,14 @@ async fn catalog_bootstrap_against_real_daemon() { .await .expect("initial snapshot"); - let snap = store.snapshot(); - assert_eq!(snap.daemons.len(), 1, "self-daemon snapshot"); - assert_eq!(snap.daemons[0].daemon_id, "konishi"); - assert!(snap.sessions.is_empty()); + assert!(store.snapshot().sessions.is_empty()); } #[tokio::test(flavor = "multi_thread")] async fn post_session_propagates_through_events_ws() { - let (_registry, daemon) = bring_up().await; - let ws_url = format!("{}/api/events", ws_base(&daemon)); - let http_base = http_base(&daemon); + let daemon = Daemon::spawn().await; + let ws_url = format!("{}/api/events", daemon.ws_base()); + let http_base = daemon.http_base(); let store = CatalogStore::default(); let bg = store.clone(); @@ -86,7 +67,6 @@ async fn post_session_propagates_through_events_ws() { let cwd = PathBuf::from(daemon.home()); let req = CreateSessionRequest { - daemon_id: "konishi".into(), cwd, command: "bash".into(), env: Default::default(), @@ -110,15 +90,14 @@ async fn post_session_propagates_through_events_ws() { let snap = store.snapshot(); assert_eq!(snap.sessions.len(), 1); - assert_eq!(snap.sessions[0].daemon_id, "konishi"); assert_eq!(snap.sessions[0].command, "bash"); } #[tokio::test(flavor = "multi_thread")] async fn delete_session_propagates_removal() { - let (_registry, daemon) = bring_up().await; - let ws_url = format!("{}/api/events", ws_base(&daemon)); - let http_base = http_base(&daemon); + let daemon = Daemon::spawn().await; + let ws_url = format!("{}/api/events", daemon.ws_base()); + let http_base = daemon.http_base(); let store = CatalogStore::default(); let bg = store.clone(); @@ -130,7 +109,6 @@ async fn delete_session_propagates_removal() { let cwd = PathBuf::from(daemon.home()); let req = CreateSessionRequest { - daemon_id: "konishi".into(), cwd, command: "bash".into(), env: Default::default(), diff --git a/crates/maestro-tui/src/sidebar.rs b/crates/maestro-tui/src/sidebar.rs index e695ea7..d17c12e 100644 --- a/crates/maestro-tui/src/sidebar.rs +++ b/crates/maestro-tui/src/sidebar.rs @@ -119,10 +119,7 @@ mod tests { #[test] fn renders_session_names() { let sb = Sidebar { - sessions: vec![ - session_named(1, "claude"), - session_named(2, "bash"), - ], + sessions: vec![session_named(1, "claude"), session_named(2, "bash")], highlight: None, attached: Default::default(), };