maestro/crates/maestro-daemon/tests/local_api.rs

//! 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, BrokerHandle, CatalogMirror, LocalEventBus};
use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, 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,
    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);
    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();
    let state = local_api::state_for(
        manager.clone(),
        hostname,
        daemon_url,
        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();
    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,
        catalog,
        _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_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 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_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 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}"
    );
}