//! 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.
//!
//! These tests require `tmux` and `git` on `$PATH`.

#![allow(clippy::result_large_err)]

use std::collections::HashMap;
use std::process::Command;
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_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<Daemon>, 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, &registry.http_base(), TOKEN).await;
        daemons.push(d);
    }
    let client = TestClient::new(&registry.http_base(), &registry.ws_base());
    client.bootstrap().await.expect("bootstrap");

    // Wait until both daemons have connected and announced themselves.
    let expected: Vec<String> = 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)
}

fn create_req(daemon_id: &str, cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
    CreateSessionRequest {
        daemon_id: daemon_id.into(),
        cwd: cwd.to_path_buf(),
        command: command.into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: None,
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn lists_sessions_from_both_daemons() {
    let (_reg, _daemons, client) = bring_up(&["alpha", "beta"]).await;
    let cwd = TempDir::new().unwrap();

    let a = client
        .spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
        .await
        .expect("spawn alpha");
    let b = client
        .spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
        .await
        .expect("spawn beta");

    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"));

    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<maestro_protocol::DaemonInfo> = 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<maestro_protocol::Session> = 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<maestro_protocol::DaemonInfo> = 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<maestro_protocol::Session> = 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 cwd = TempDir::new().unwrap();

    let s = client
        .spawn_session(&create_req("solo", 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");

    // Drive input through client A; both attachments should observe it via
    // the daemon's PTY broadcast.
    let send = b"echo MAESTRO_E2E_SHARED\n".to_vec();
    a.send(WsMessage::Binary(send)).await.expect("send keys");

    let want = b"MAESTRO_E2E_SHARED";
    let deadline = Instant::now() + Duration::from_secs(5);
    let mut buf_a = Vec::new();
    let mut buf_b = Vec::new();
    while Instant::now() < deadline {
        tokio::select! {
            msg = a.next() => match msg {
                Some(Ok(WsMessage::Binary(bytes))) => buf_a.extend(bytes),
                Some(Ok(_)) => {}
                _ => break,
            },
            msg = b.next() => match msg {
                Some(Ok(WsMessage::Binary(bytes))) => buf_b.extend(bytes),
                Some(Ok(_)) => {}
                _ => break,
            },
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
        if buf_a.windows(want.len()).any(|w| w == want)
            && buf_b.windows(want.len()).any(|w| w == want)
        {
            break;
        }
    }
    assert!(
        buf_a.windows(want.len()).any(|w| w == want),
        "client A never saw the echo"
    );
    assert!(
        buf_b.windows(want.len()).any(|w| w == want),
        "client B never saw the echo"
    );

    let _ = a.close(None).await;
    let _ = b.close(None).await;
    client.kill_session(s.id).await.expect("kill");
}

#[tokio::test(flavor = "multi_thread")]
async fn cleanup_on_kill_removes_worktree() {
    let (_reg, _daemons, client) = bring_up(&["host"]).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(),
        metadata: HashMap::new(),
        worktree: Some(WorktreeSpec {
            base_repo: base.path().to_path_buf(),
            branch: "feature/e2e".into(),
            cleanup_on_kill: true,
        }),
    };

    let s = client.spawn_session(&req).await.expect("spawn worktree");
    // The daemon's WorktreeManager puts new worktrees under
    // `<state_dir>/worktrees/<branch>`. We can't peek the daemon's state dir
    // directly from here, but we can verify by asking git: a worktree must
    // exist that lives outside the original repo.
    let worktrees = git_worktree_list(base.path());
    assert!(
        worktrees.iter().any(|w| w.contains("feature/e2e")),
        "expected a worktree for feature/e2e branch, got: {worktrees:?}"
    );

    client.kill_session(s.id).await.expect("kill");

    // Give the daemon a moment to remove the worktree, then verify.
    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        let after = git_worktree_list(base.path());
        if !after.iter().any(|w| w.contains("feature/e2e")) {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!(
        "worktree feature/e2e still present after kill: {:?}",
        git_worktree_list(base.path())
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn daemon_restart_recovers_session() {
    let (_reg, mut daemons, client) = bring_up(&["restarter"]).await;
    let cwd = TempDir::new().unwrap();

    let s = client
        .spawn_session(&create_req("restarter", 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;

    wait_for_async(
        "session reappears in catalog",
        Duration::from_secs(10),
        || {
            let client = client.clone();
            let id = s.id;
            async move {
                client
                    .list_sessions()
                    .await
                    .map(|ss| ss.iter().any(|x| x.id == id))
                    .unwrap_or(false)
            }
        },
    )
    .await;

    client.kill_session(s.id).await.expect("kill");
}

#[tokio::test(flavor = "multi_thread")]
async fn concurrent_spawns_each_get_unique_ids() {
    let (_reg, _daemons, client) = bring_up(&["parallel"]).await;
    let cwd = TempDir::new().unwrap();

    let mut handles = Vec::new();
    for _ in 0..5 {
        let client = client.clone();
        let path = cwd.path().to_path_buf();
        handles.push(tokio::spawn(async move {
            client
                .spawn_session(&create_req("parallel", &path, "sleep 30"))
                .await
                .expect("spawn")
        }));
    }
    let mut ids = Vec::new();
    for h in handles {
        ids.push(h.await.unwrap().id);
    }
    ids.sort();
    ids.dedup();
    assert_eq!(ids.len(), 5, "expected 5 distinct session ids");

    for id in ids {
        client.kill_session(id).await.expect("kill");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn spawn_records_history_and_infers_gitroot_for_form() {
    // #24/#25/#26/#28: a real spawn writes a durable spawn-history record, which
    // 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);

    // A git repo as the working dir, so gitroot has a repo to infer.
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let cwd = repo.path().to_string_lossy().to_string();

    // 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"))
        .await
        .expect("spawn");

    // The recents picker reads GET /recent-dirs on open: the just-spawned dir
    // must surface, tagged with the command it was last used with. The history
    // write is best-effort right after the spawn, so poll briefly.
    wait_for_async(
        "recent-dirs surfaces the spawned cwd",
        Duration::from_secs(5),
        || {
            let url = format!("{daemon_base}/api/daemons/host/recent-dirs");
            let want = cwd.clone();
            async move {
                let resp = match reqwest::get(&url).await {
                    Ok(r) => r,
                    Err(_) => return false,
                };
                let recents: maestro_protocol::RecentDirs = match resp.json().await {
                    Ok(v) => v,
                    Err(_) => return false,
                };
                recents
                    .dirs
                    .iter()
                    .any(|d| d.cwd.to_string_lossy() == want && d.last_command == "sleep 30")
            }
        },
    )
    .await;

    // 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"))
        .query(&[("path", cwd.as_str())])
        .send()
        .await
        .expect("gitroot request")
        .json()
        .await
        .expect("gitroot json");
    assert_eq!(
        gr.root.as_deref().map(|p| p.to_string_lossy().to_string()),
        Some(cwd.clone()),
        "gitroot should infer the enclosing repo"
    );

    client.kill_session(s.id).await.expect("kill");
}

#[tokio::test(flavor = "multi_thread")]
async fn activity_reflects_foreground_and_input_blocking() {
    // NAP-340 (#18): the watcher infers each session's foreground command and
    // activity from structured tmux/proc state. `cat` parks in read() on the
    // pty → AwaitingInput; `sleep` is quiescent but not reading the tty → Idle
    // (the distinction the /proc/<pid>/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 cwd = TempDir::new().unwrap();

    let cat = client
        .spawn_session(&create_req("host", cwd.path(), "cat"))
        .await
        .expect("spawn cat");
    let slp = client
        .spawn_session(&create_req("host", cwd.path(), "sleep 300"))
        .await
        .expect("spawn sleep");

    // Initial state is Active; the watcher needs a couple ticks past the
    // quiescence window (3s) to settle into the steady states.
    wait_for_async(
        "cat → AwaitingInput, sleep → Idle, with correct foreground",
        Duration::from_secs(20),
        || {
            let client = client.clone();
            let (cat_id, sleep_id) = (cat.id, slp.id);
            async move {
                let sessions = match client.list_sessions().await {
                    Ok(s) => s,
                    Err(_) => return false,
                };
                let cat_ok = sessions
                    .iter()
                    .find(|s| s.id == cat_id)
                    .map(|s| {
                        s.activity == ActivityState::AwaitingInput
                            && s.foreground.as_deref() == Some("cat")
                    })
                    .unwrap_or(false);
                let sleep_ok = sessions
                    .iter()
                    .find(|s| s.id == sleep_id)
                    .map(|s| {
                        s.activity == ActivityState::Idle
                            && s.foreground.as_deref() == Some("sleep")
                    })
                    .unwrap_or(false);
                cat_ok && sleep_ok
            }
        },
    )
    .await;

    client.kill_session(cat.id).await.expect("kill cat");
    client.kill_session(slp.id).await.expect("kill sleep");
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn init_repo(dir: &std::path::Path) {
    run_git(&["init", "-q", "-b", "main"], dir);
    run_git(&["config", "user.email", "e2e@example.com"], dir);
    run_git(&["config", "user.name", "E2E"], dir);
    std::fs::write(dir.join("README"), b"hello\n").unwrap();
    run_git(&["add", "."], dir);
    run_git(&["commit", "-q", "-m", "initial"], dir);
}

fn run_git(args: &[&str], cwd: &std::path::Path) {
    let out = Command::new("git")
        .arg("-C")
        .arg(cwd)
        .args(args)
        .output()
        .expect("git spawn");
    assert!(
        out.status.success(),
        "git {args:?} failed: {}",
        String::from_utf8_lossy(&out.stderr)
    );
}

fn git_worktree_list(dir: &std::path::Path) -> Vec<String> {
    let out = Command::new("git")
        .arg("-C")
        .arg(dir)
        .args(["worktree", "list", "--porcelain"])
        .output()
        .expect("git worktree list");
    String::from_utf8_lossy(&out.stdout)
        .lines()
        .map(str::to_string)
        .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",
        &registry.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",
        &registry.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(&registry.http_base(), &registry.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<maestro_protocol::Session> = 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",
        &registry.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",
        &registry.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(&registry.http_base(), &registry.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<maestro_protocol::Session> = 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
    // Snapshot, then a SessionAdded after we spawn a session via REST.
    let (_reg, daemons, client) = bring_up(&["alpha"]).await;
    let cwd = TempDir::new().unwrap();

    let alpha_addr = daemons[0].bind_addr;
    let ws_url = format!("ws://{alpha_addr}/api/events");
    let (mut ws, _) = tokio_tungstenite::connect_async(ws_url)
        .await
        .expect("events WS handshake");

    // 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"));
            assert!(sessions.is_empty());
        }
        other => panic!("expected initial Snapshot, got {other:?}"),
    }

    let session = client
        .spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
        .await
        .expect("spawn alpha");

    // 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 }
                    if s.id == session.id =>
                {
                    return true;
                }
                _ => continue,
            }
        }
    })
    .await
    .unwrap_or(false);
    assert!(
        saw_added,
        "events WS never delivered SessionAdded for own session"
    );

    client.kill_session(session.id).await.expect("kill");
}

// ---------------------------------------------------------------------------
// NAP-244 — CLI rewired through the home-daemon broker, with registry fallback
// ---------------------------------------------------------------------------

/// Run the real `maestro` binary with an isolated env and capture its output.
fn run_cli(args: &[&str], envs: &[(&str, &str)]) -> std::process::Output {
    let bin = maestro_e2e::workspace_target_bin("maestro", "maestro");
    let mut cmd = Command::new(&bin);
    cmd.args(args);
    cmd.env_clear().env(
        "PATH",
        std::env::var("PATH").unwrap_or_else(|_| "/usr/bin:/bin".into()),
    );
    for (k, v) in envs {
        cmd.env(k, v);
    }
    cmd.output().expect("run maestro cli")
}

#[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;
    let cwd = TempDir::new().unwrap();
    let session = client
        .spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
        .await
        .expect("spawn");

    let daemon_url = format!("http://{}", daemons[0].bind_addr);
    let home = TempDir::new().unwrap();
    let cfg_path = home.path().join("config.toml");
    let out = run_cli(
        &["list", "--json"],
        &[
            ("HOME", home.path().to_str().unwrap()),
            ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()),
            ("MAESTRO_HOME_DAEMON_URL", &daemon_url),
        ],
    );
    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 home daemon: {stdout}"
    );
    assert!(
        !stderr.contains("falling back"),
        "unexpected registry fallback: {stderr}"
    );

    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");

    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"],
        &[
            ("HOME", home.path().to_str().unwrap()),
            ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()),
            ("MAESTRO_HOME_DAEMON_URL", &dead_daemon),
            ("MAESTRO_REGISTRY_URL", &registry_url),
        ],
    );
    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();
}

async fn next_event<S>(ws: &mut S) -> maestro_protocol::RegistryEvent
where
    S: StreamExt<Item = Result<WsMessage, tokio_tungstenite::tungstenite::Error>> + Unpin,
{
    loop {
        match ws.next().await {
            Some(Ok(WsMessage::Text(t))) => {
                return serde_json::from_str(&t).expect("RegistryEvent")
            }
            Some(Ok(_)) => continue,
            other => panic!("WS closed waiting for event: {other:?}"),
        }
    }
}