feat(maestro): daemon local-view HTTP API + broker WS self-short-circuit (NAP-243)
ci / rust (push) Failing after 10m36s
ci / rust (push) Failing after 10m36s
New module crates/maestro-daemon/src/local_api.rs serving:
* GET /api/sessions — own sessions
* GET /api/daemons — self only (v1; peer mirror is NAP-246)
* GET /api/web/info — daemon identity (browser header data)
* POST /api/sessions — create on self; 501 for cross-daemon (NAP-247)
* DELETE /api/sessions/:id — kill on self; 404 for unknown
* WS /api/attach/:id — delegate to existing PTY subscribe when self-owned,
404 otherwise (cross-daemon broker is NAP-247)
The new routes share bind_addr with the legacy /sessions/:id/stream WS
endpoint, which keeps working unchanged. The shared serve loop is exposed
as attach::serve_local so the broker WS can reuse the PTY pump.
Trust model documented at the top of the module: no JWT, no cookie on
the new routes — same trust as tmux being one-user-wide. Operators who
bind a perimeter address opt into one-segment-wide trust.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
//! 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.
|
||||
|
||||
mod common;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::net::{Ipv4Addr, SocketAddr};
|
||||
use std::time::Duration;
|
||||
|
||||
use common::DaemonFixture;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use maestro_daemon::local_api;
|
||||
use maestro_protocol::{CreateSessionRequest, DaemonInfo, Session};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
|
||||
use uuid::Uuid;
|
||||
|
||||
type WsClient = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
|
||||
|
||||
struct Harness {
|
||||
addr: SocketAddr,
|
||||
manager: maestro_daemon::SessionManager,
|
||||
_shutdown: oneshot::Sender<()>,
|
||||
_fx: DaemonFixture,
|
||||
}
|
||||
|
||||
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 router = local_api::router(state).merge(local_api::broker_router(manager.clone()));
|
||||
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
|
||||
.await
|
||||
.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let (tx, rx) = oneshot::channel::<()>();
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(listener, router)
|
||||
.with_graceful_shutdown(async move {
|
||||
let _ = rx.await;
|
||||
})
|
||||
.await;
|
||||
});
|
||||
Harness {
|
||||
addr,
|
||||
manager,
|
||||
_shutdown: tx,
|
||||
_fx: fx,
|
||||
}
|
||||
}
|
||||
|
||||
fn url(addr: SocketAddr, path: &str) -> String {
|
||||
format!("http://{addr}{path}")
|
||||
}
|
||||
|
||||
#[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"))
|
||||
.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!(resp["version"].is_string());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_daemons_returns_self_only() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
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");
|
||||
}
|
||||
|
||||
#[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 empty: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(empty.is_empty());
|
||||
|
||||
let req = CreateSessionRequest {
|
||||
daemon_id: "laptop".into(),
|
||||
cwd: h.manager.daemon_id().into(),
|
||||
command: "sleep 60".into(),
|
||||
env: HashMap::new(),
|
||||
metadata: HashMap::new(),
|
||||
worktree: None,
|
||||
};
|
||||
let created: Session = reqwest::Client::new()
|
||||
.post(url(h.addr, "/api/sessions"))
|
||||
.json(&req)
|
||||
.send()
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(created.daemon_id, "laptop");
|
||||
|
||||
let after: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
|
||||
.await
|
||||
.unwrap()
|
||||
.json()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(after.len(), 1);
|
||||
assert_eq!(after[0].id, created.id);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn post_to_peer_daemon_id_returns_501_not_proxy() {
|
||||
// Cross-daemon broker is NAP-247; until it lands a daemon must refuse
|
||||
// (rather than silently misroute) a create targeting a peer.
|
||||
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::NOT_IMPLEMENTED);
|
||||
let body: serde_json::Value = resp.json().await.unwrap();
|
||||
assert_eq!(body["error"]["code"], "remote_create_unsupported");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_unknown_session_returns_404() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
let resp = reqwest::Client::new()
|
||||
.delete(url(h.addr, &format!("/api/sessions/{}", Uuid::new_v4())))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_owned_session_kills_it() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
let session = 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();
|
||||
let resp = reqwest::Client::new()
|
||||
.delete(url(h.addr, &format!("/api/sessions/{}", session.id)))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
|
||||
assert!(!h.manager.has(session.id).await);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn broker_ws_attaches_to_local_session() {
|
||||
let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await;
|
||||
let session = h
|
||||
.manager
|
||||
.create(CreateSessionRequest {
|
||||
daemon_id: "laptop".into(),
|
||||
cwd: "/tmp".into(),
|
||||
command: "bash --noprofile --norc".into(),
|
||||
env: HashMap::new(),
|
||||
metadata: HashMap::new(),
|
||||
worktree: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let ws_url = format!("ws://{}/api/attach/{}", h.addr, session.id);
|
||||
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()))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let saw_hello = tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
match ws.next().await {
|
||||
Some(Ok(WsMessage::Binary(bytes))) => {
|
||||
if String::from_utf8_lossy(&bytes).contains("hello-broker") {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
Some(Ok(_)) => continue,
|
||||
Some(Err(_)) | None => return false,
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap_or(false);
|
||||
assert!(saw_hello, "PTY did not echo through the broker 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;
|
||||
let ws_url = format!("ws://{}/api/attach/{}", h.addr, Uuid::new_v4());
|
||||
let err = connect_async(ws_url)
|
||||
.await
|
||||
.expect_err("expected handshake fail");
|
||||
let msg = err.to_string();
|
||||
assert!(
|
||||
msg.contains("404") || msg.to_lowercase().contains("not found"),
|
||||
"{msg}"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user