//! Cross-process end-to-end tests for the headless Maestro slice.
//!
//! 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`.

#![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, TestClient};
use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
use tempfile::TempDir;
use tokio_tungstenite::tungstenite::Message as WsMessage;

async fn bring_up() -> (Daemon, TestClient) {
    let daemon = Daemon::spawn().await;
    let client = TestClient::new(&daemon.http_base());
    (daemon, client)
}

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

/// A worktree spec built through its wire shape. The daemon and clients ship
/// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
/// object carries only the fields the lifecycle actually honours.
fn worktree_spec(base_repo: &std::path::Path, branch: &str, create_branch: bool) -> WorktreeSpec {
    WorktreeSpec {
        base_repo: base_repo.to_path_buf(),
        branch: branch.into(),
        create_branch,
    }
}

/// Spawn request carrying a worktree spec against `base`.
fn worktree_req(base: &std::path::Path, branch: &str, create_branch: bool) -> CreateSessionRequest {
    CreateSessionRequest {
        cwd: base.to_path_buf(),
        command: "sleep 30".into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: Some(worktree_spec(base, branch, create_branch)),
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn lists_spawned_sessions() {
    let (_daemon, client) = bring_up().await;
    let cwd = TempDir::new().unwrap();

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

    let sessions = client.list_sessions().await.expect("list");
    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 multi_attach_sees_same_output() {
    let (_daemon, client) = bring_up().await;
    let cwd = TempDir::new().unwrap();

    let s = client
        .spawn_session(&create_req(cwd.path(), "bash --noprofile --norc"))
        .await
        .expect("spawn");

    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.
    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 kill_removes_clean_worktree() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());

    let req = CreateSessionRequest {
        cwd: base.path().to_path_buf(),
        command: "sleep 30".into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: Some(worktree_spec(base.path(), "feature/e2e", true)),
    };

    let s = client.spawn_session(&req).await.expect("spawn worktree");
    // Worktrees live inside the base repo at `<repo>/worktrees/<branch>`
    // (#56), so `git worktree list` must show the predictable in-repo path.
    let worktrees = git_worktree_list(base.path());
    assert!(
        worktrees
            .iter()
            .any(|w| w.contains("worktrees/feature/e2e")),
        "expected a worktree for feature/e2e under <repo>/worktrees, got: {worktrees:?}"
    );

    let _origin = throwaway_origin(base.path());
    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())
    );
}

/// The kill guard (#79): killing a session whose worktree holds uncommitted
/// changes is refused outright. Nothing happens — the session keeps running
/// as the operator's handle on their unfinished work, the tree stays on disk,
/// and the refusal names what blocked it and how much.
#[tokio::test(flavor = "multi_thread")]
async fn kill_refuses_worktree_with_uncommitted_changes() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let wt_path = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("unsaved");

    let req = CreateSessionRequest {
        cwd: base.path().to_path_buf(),
        command: "sleep 30".into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: Some(worktree_spec(base.path(), "feature/unsaved", true)),
    };

    let s = client.spawn_session(&req).await.expect("spawn worktree");
    assert!(wt_path.exists());
    std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();

    let err = client
        .kill_session(s.id)
        .await
        .expect_err("kill must refuse");
    match err {
        maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. }) => {
            assert_eq!(info.code, "kill_guarded", "{info:?}");
            let breakdown = info.unsaved_work.expect("breakdown on the wire");
            assert_eq!(breakdown.uncommitted_files, 1);
        }
        other => panic!("expected a kill_guarded refusal, got {other:?}"),
    }

    // The session survives and the dirty worktree stays findable on disk.
    let sessions = client.list_sessions().await.unwrap_or_default();
    assert!(
        sessions.iter().any(|x| x.id == s.id),
        "a refused kill must keep the session alive"
    );
    assert!(
        wt_path.exists(),
        "guarded worktree must be left on disk after kill"
    );
    // Left findable, not silently half-deleted: git still registers it and
    // the uncommitted edit survives byte for byte (#66).
    assert!(
        git_worktree_list(base.path())
            .iter()
            .any(|w| w.contains("feature/unsaved")),
        "guarded worktree must stay registered in git worktree list"
    );
    assert_eq!(
        std::fs::read(wt_path.join("README")).unwrap(),
        b"unsaved edits\n",
        "uncommitted changes must survive the guarded kill"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn spawn_on_existing_branch_checks_it_out() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    run_git(&["branch", "feature/existing"], base.path());

    let req = CreateSessionRequest {
        cwd: base.path().to_path_buf(),
        command: "sleep 30".into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: Some(worktree_spec(base.path(), "feature/existing", false)),
    };

    let s = client
        .spawn_session(&req)
        .await
        .expect("spawn on existing branch");
    let worktrees = git_worktree_list(base.path());
    assert!(
        worktrees.iter().any(|w| w.contains("feature/existing")),
        "expected a worktree checked out at feature/existing, got: {worktrees:?}"
    );

    let _origin = throwaway_origin(base.path());
    client.kill_session(s.id).await.expect("kill");
}

/// #66: a fresh branch spawns a worktree at `<repo>/worktrees/<branch>` and
/// the session actually runs inside it — `current_cwd` comes from tmux's live
/// pane state, not from the spawn request.
#[tokio::test(flavor = "multi_thread")]
async fn spawn_new_branch_creates_worktree_and_runs_in_it() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let wt_path = base.path().join("worktrees").join("feature").join("fresh");

    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/fresh", true))
        .await
        .expect("spawn new-branch worktree");

    assert_eq!(
        s.cwd, wt_path,
        "session must report the worktree as its cwd"
    );
    assert!(wt_path.is_dir(), "worktree directory must exist on disk");

    // The activity watcher fills `current_cwd` from tmux's pane_current_path,
    // so this is the daemon confirming where the process really runs.
    wait_for_async(
        "session's live cwd is the worktree",
        Duration::from_secs(20),
        || {
            let client = client.clone();
            let id = s.id;
            let want = wt_path.clone();
            async move {
                client
                    .list_sessions()
                    .await
                    .ok()
                    .and_then(|ss| ss.iter().find(|s| s.id == id).cloned())
                    .map(|s| s.current_cwd == Some(want.clone()))
                    .unwrap_or(false)
            }
        },
    )
    .await;

    let _origin = throwaway_origin(base.path());
    client.kill_session(s.id).await.expect("kill");
}

/// #66: a second spawn onto a branch that already has a worktree is rejected
/// with a structured worktree error; the first worktree stays intact.
#[tokio::test(flavor = "multi_thread")]
async fn second_spawn_on_same_branch_is_rejected() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());

    let first = client
        .spawn_session(&worktree_req(base.path(), "feature/dup", true))
        .await
        .expect("first spawn");
    let dup = client
        .spawn_session(&worktree_req(base.path(), "feature/dup", true))
        .await;
    match dup {
        Err(maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. })) => {
            assert_eq!(
                info.code, "worktree_failed",
                "expected worktree error: {info:?}"
            );
        }
        other => panic!("expected rejection for duplicate branch, got {other:?}"),
    }

    // Exactly one registration, one directory, still usable.
    // Count branch records only: the worktree path now contains the branch
    // name too (`worktrees/feature/dup`), so a substring match over every line
    // would count the same tree twice.
    let holders = git_worktree_list(base.path())
        .iter()
        .filter(|w| w.as_str() == "branch refs/heads/feature/dup")
        .count();
    assert_eq!(holders, 1, "duplicate spawn must not register another tree");
    assert!(
        base.path()
            .join("worktrees")
            .join("feature")
            .join("dup")
            .is_dir(),
        "original worktree must be untouched"
    );

    let _origin = throwaway_origin(base.path());
    client.kill_session(first.id).await.expect("kill");
}

/// #66/#56: when the client picks a *linked worktree* as base_repo, placement
/// resolves through `--git-common-dir` to the main checkout, so the new tree
/// lands at `<main>/worktrees/<branch>` instead of nesting inside another
/// worktree.
#[tokio::test(flavor = "multi_thread")]
async fn spawn_from_linked_worktree_places_in_main_checkout() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let linked = base.path().join("linked");
    run_git(
        &[
            "worktree",
            "add",
            "-q",
            "-b",
            "other",
            &linked.to_string_lossy(),
        ],
        base.path(),
    );

    let s = client
        .spawn_session(&worktree_req(&linked, "feature/nested", true))
        .await
        .expect("spawn from linked worktree");

    let want = base.path().join("worktrees").join("feature").join("nested");
    assert_eq!(
        s.cwd, want,
        "worktree must be placed under the main checkout"
    );
    assert!(want.is_dir());
    assert!(
        !s.cwd.starts_with(&linked),
        "worktrees must never nest inside another worktree"
    );

    let _origin = throwaway_origin(base.path());
    client.kill_session(s.id).await.expect("kill");
}

// ---------------------------------------------------------------------------
// Worktree lifecycle: collisions, retention, reclaim (#88)
// ---------------------------------------------------------------------------

/// #77/#88: the target path `<repo>/worktrees/<branch>` is occupied by a real
/// registered worktree the operator made by hand. The daemon sidesteps to the
/// suffixed directory and never touches the occupant — and both trees keep
/// working: two registered worktrees, two sessions, each running in its own.
#[tokio::test(flavor = "multi_thread")]
async fn occupied_path_gets_suffixed_directory_and_both_worktrees_work() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());

    // An operator-made worktree sits exactly where `feature/collide` would go.
    // Its branch is a different one, so the requested branch stays creatable.
    let squatter = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("collide");
    run_git(
        &[
            "worktree",
            "add",
            "-q",
            "-b",
            "operator/manual",
            &squatter.to_string_lossy(),
        ],
        base.path(),
    );

    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/collide", true))
        .await
        .expect("spawn onto occupied path");

    let suffixed = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("collide-2");
    assert_eq!(
        s.cwd, suffixed,
        "the daemon must take the suffixed directory, not the occupied one"
    );
    assert!(suffixed.is_dir(), "suffixed worktree must exist on disk");

    // Both trees are registered, distinct, and the squatter is byte-intact.
    let registrations = git_worktree_list(base.path());
    assert!(
        registrations
            .iter()
            .any(|l| l == &format!("worktree {}", squatter.canonicalize().unwrap().display())),
        "operator's worktree must stay registered: {registrations:?}"
    );
    assert!(
        registrations.iter().any(|l| l.contains("collide-2")),
        "suffixed worktree must be registered too: {registrations:?}"
    );

    // Both sessions run inside their own tree at the same time.
    let sessions = client.list_sessions().await.expect("list");
    assert!(
        sessions.iter().any(|x| x.id == s.id && x.cwd == suffixed),
        "session must keep running in its suffixed tree"
    );

    let _origin = throwaway_origin(base.path());
    client.kill_session(s.id).await.expect("kill");

    // Only the daemon-made trees go; the operator's manual one survives.
    let after = git_worktree_list(base.path());
    assert!(
        after
            .iter()
            .any(|l| l == &format!("worktree {}", squatter.canonicalize().unwrap().display())),
        "operator's worktree must outlive the sessions: {after:?}"
    );
}

/// #77/#88: the target path holds a plain directory the daemon never made and
/// git has never heard of. The daemon neither adopts it nor errors — it builds
/// beside it and leaves every byte of it alone.
#[tokio::test(flavor = "multi_thread")]
async fn foreign_directory_is_sidestepped_never_adopted() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());

    let squatter = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("foreign");
    std::fs::create_dir_all(&squatter).unwrap();
    std::fs::write(squatter.join("operator-notes.txt"), b"mine\n").unwrap();

    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/foreign", true))
        .await
        .expect("spawn must sidestep, not adopt or fail");

    let suffixed = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("foreign-2");
    assert_eq!(s.cwd, suffixed, "session must run in the suffixed tree");
    assert_eq!(
        std::fs::read(squatter.join("operator-notes.txt")).unwrap(),
        b"mine\n",
        "the foreign directory must be untouched"
    );
    let registrations = git_worktree_list(base.path());
    let registered_paths: Vec<&str> = registrations
        .iter()
        .filter_map(|l| l.strip_prefix("worktree "))
        .collect();
    assert!(
        !registered_paths
            .iter()
            .any(|p| *p == squatter.canonicalize().unwrap().display().to_string()),
        "a plain directory must never be adopted as a worktree: {registered_paths:?}"
    );

    let _origin = throwaway_origin(base.path());
    client.kill_session(s.id).await.expect("kill");
    assert!(squatter.join("operator-notes.txt").exists());
}

/// #79/#88: killing with uncommitted changes is refused; committing and
/// pushing that work clears the guard and the very same kill then succeeds
/// and removes the tree. The session is the handle on the work: while it is
/// unsaved the session stays, once it is saved the tree goes.
#[tokio::test(flavor = "multi_thread")]
async fn guarded_kill_then_commit_and_push_makes_the_same_kill_succeed() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let _origin = throwaway_origin(base.path());

    let wt = base.path().join("worktrees").join("feature").join("guard");
    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/guard", true))
        .await
        .expect("spawn");
    std::fs::write(wt.join("README"), b"work in progress\n").unwrap();

    // Refused: the edit is uncommitted.
    let err = client
        .kill_session(s.id)
        .await
        .expect_err("kill must refuse");
    match err {
        maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. }) => {
            assert_eq!(info.code, "kill_guarded", "{info:?}");
            let breakdown = info.unsaved_work.expect("breakdown on the wire");
            assert_eq!(breakdown.uncommitted_files, 1);
            assert_eq!(breakdown.unpushed_commits, 0);
        }
        other => panic!("expected a kill_guarded refusal, got {other:?}"),
    }
    assert!(
        client
            .list_sessions()
            .await
            .unwrap()
            .iter()
            .any(|x| x.id == s.id),
        "a refused kill keeps the session running"
    );
    assert_eq!(
        std::fs::read(wt.join("README")).unwrap(),
        b"work in progress\n",
        "the worktree is untouched by the refusal"
    );

    // The operator does exactly what the refusal asked for: save and push.
    commit_all(&wt, "wip");
    push_all(base.path());

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

    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if !wt.exists() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    assert!(
        !wt.exists(),
        "a cleared worktree must be removed after kill"
    );
    let registrations = git_worktree_list(base.path());
    assert!(
        !registrations.iter().any(|l| l.contains("feature/guard")),
        "removal must deregister the tree: {registrations:?}"
    );
}

/// #81/#88: commits that no remote has refuse the kill; `--force` ends the
/// session anyway but RETAINS the worktree — force never destroys blocked
/// work, it parks it in the retained store with reason `force-killed`.
#[tokio::test(flavor = "multi_thread")]
async fn unpushed_commits_block_kill_and_force_retains_the_tree() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let _origin = throwaway_origin(base.path());

    let wt = base.path().join("worktrees").join("feature").join("local");
    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/local", true))
        .await
        .expect("spawn");
    std::fs::write(wt.join("README"), b"local only\n").unwrap();
    commit_all(&wt, "not pushed yet");

    // Refused: the commit is on no remote-tracking ref.
    let err = client
        .kill_session(s.id)
        .await
        .expect_err("kill must refuse");
    match err {
        maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. }) => {
            assert_eq!(info.code, "kill_guarded", "{info:?}");
            let breakdown = info.unsaved_work.expect("breakdown on the wire");
            assert_eq!(breakdown.unpushed_commits, 1);
            assert_eq!(breakdown.uncommitted_files, 0);
        }
        other => panic!("expected a kill_guarded refusal, got {other:?}"),
    }

    // Force overrides the refusal and nothing else: session gone, tree kept.
    let outcome = client.kill_session_forced(s.id).await.expect("forced kill");
    let retained = outcome
        .retained
        .expect("force must retain the guarded tree");
    assert_eq!(
        retained.reason,
        maestro_protocol::RetainedReason::ForceKilled
    );
    assert_eq!(retained.path, wt);
    assert_eq!(retained.branch, "feature/local");
    assert_eq!(retained.in_the_way.unpushed_commits, 1);

    assert!(
        !client
            .list_sessions()
            .await
            .unwrap()
            .iter()
            .any(|x| x.id == s.id),
        "the forced kill ends the session"
    );
    assert!(wt.is_dir(), "a force-retained worktree must stay on disk");
    assert!(
        git_worktree_list(base.path())
            .iter()
            .any(|l| l.contains("feature/local")),
        "the retained tree stays registered in git"
    );
}

/// #82/#88: a session that exits on its own never reaches the kill path, but
/// its worktree follows the same rule. Dirty → retained with reason
/// `exited-with-unsaved-work`; pristine → removed, leaving nothing behind.
#[tokio::test(flavor = "multi_thread")]
async fn natural_exit_settles_dirty_and_pristine_worktrees_differently() {
    let (_daemon, client) = bring_up().await;

    // Two repos so the two cases cannot see each other's state.
    let dirty_repo = TempDir::new().unwrap();
    init_repo(dirty_repo.path());
    let _origin = throwaway_origin(dirty_repo.path());
    let pristine_repo = TempDir::new().unwrap();
    init_repo(pristine_repo.path());
    let _origin_pristine = throwaway_origin(pristine_repo.path());

    let dirty_wt = dirty_repo
        .path()
        .join("worktrees")
        .join("feature")
        .join("messy");
    let pristine_wt = pristine_repo
        .path()
        .join("worktrees")
        .join("feature")
        .join("tidy");

    let _dirty = client
        .spawn_session(&CreateSessionRequest {
            cwd: dirty_repo.path().to_path_buf(),
            command: "sh -c 'echo unsaved > scratch.txt'".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(worktree_spec(dirty_repo.path(), "feature/messy", true)),
        })
        .await
        .expect("spawn dirty");
    let _pristine = client
        .spawn_session(&create_req_cmd(
            pristine_repo.path(),
            "true",
            "feature/tidy",
        ))
        .await
        .expect("spawn pristine");

    assert!(dirty_wt.is_dir() && pristine_wt.is_dir());

    // The watcher reaps dead tmux sessions every couple of ticks and settles
    // each worktree accordingly. Poll for both outcomes.
    wait_for_async(
        "dirty exit retains its worktree",
        Duration::from_secs(20),
        || {
            let client = client.clone();
            let path = dirty_wt.clone();
            async move {
                client
                    .list_worktrees()
                    .await
                    .ok()
                    .map(|wts| {
                        wts.iter().any(|w| {
                            w.path == path
                                && w.reason
                                    == maestro_protocol::RetainedReason::ExitedWithUnsavedWork
                        })
                    })
                    .unwrap_or(false)
            }
        },
    )
    .await;
    wait_for_async(
        "pristine exit removes its worktree entirely",
        Duration::from_secs(20),
        || {
            let client = client.clone();
            let repo = pristine_repo.path().to_path_buf();
            let path = pristine_wt.clone();
            async move {
                if path.exists() {
                    return false;
                }
                let listed = client.list_worktrees().await.ok();
                let listed_clear = listed
                    .map(|wts| !wts.iter().any(|w| w.path == path))
                    .unwrap_or(false);
                let registered = git_worktree_list(&repo);
                !registered.iter().any(|l| l.contains("feature/tidy")) && listed_clear
            }
        },
    )
    .await;

    // The dirty tree is still fully there, file included.
    assert!(dirty_wt.is_dir());
    assert_eq!(
        std::fs::read(dirty_wt.join("scratch.txt")).unwrap(),
        b"unsaved\n"
    );
}

fn create_req_cmd(cwd: &std::path::Path, command: &str, branch: &str) -> CreateSessionRequest {
    CreateSessionRequest {
        cwd: cwd.to_path_buf(),
        command: command.into(),
        env: HashMap::new(),
        metadata: HashMap::new(),
        worktree: Some(worktree_spec(cwd, branch, true)),
    }
}

/// #80/#88: `GET /api/worktrees` lists what the daemon is holding, with the
/// reason and a live in-the-way breakdown; reclaim removes the directory and
/// the store entry but NEVER the branch — its commits outlive the reclaim.
#[tokio::test(flavor = "multi_thread")]
async fn worktrees_endpoint_lists_reasons_and_reclaim_keeps_the_branch() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let _origin = throwaway_origin(base.path());

    let wt = base.path().join("worktrees").join("feature").join("parked");
    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/parked", true))
        .await
        .expect("spawn");
    std::fs::write(wt.join("README"), b"worth keeping\n").unwrap();
    commit_all(&wt, "worth keeping");
    let sha = branch_sha(&wt, "feature/parked");

    let outcome = client.kill_session_forced(s.id).await.expect("force kill");
    assert!(outcome.retained.is_some());

    let listed = client.list_worktrees().await.expect("list worktrees");
    assert_eq!(listed.len(), 1, "exactly one retained entry: {listed:?}");
    let entry = &listed[0];
    assert_eq!(entry.path, wt);
    assert_eq!(entry.branch, "feature/parked");
    assert_eq!(entry.reason, maestro_protocol::RetainedReason::ForceKilled);
    assert_eq!(entry.in_the_way.unpushed_commits, 1);

    client.reclaim_worktree(entry.id).await.expect("reclaim");
    assert!(!wt.exists(), "reclaim removes the directory");
    assert!(
        client.list_worktrees().await.unwrap().is_empty(),
        "reclaim drops the store entry"
    );

    // The whole point of reclaim: the branch and its commits survive.
    assert_eq!(
        branch_sha(base.path(), "feature/parked"),
        sha,
        "the branch must keep pointing at the same commit after reclaim"
    );
}

/// The retained store is state-dir backed (#80): a daemon restart must not
/// orphan the trees it is holding — they are still on disk with entries
/// pointing at them, and still reclaimable afterwards.
#[tokio::test(flavor = "multi_thread")]
async fn retained_store_survives_a_daemon_restart() {
    let (mut daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    let _origin = throwaway_origin(base.path());

    let wt = base
        .path()
        .join("worktrees")
        .join("feature")
        .join("durable");
    let s = client
        .spawn_session(&worktree_req(base.path(), "feature/durable", true))
        .await
        .expect("spawn");
    std::fs::write(wt.join("README"), b"survives restarts\n").unwrap();
    commit_all(&wt, "survives restarts");
    client.kill_session_forced(s.id).await.expect("force kill");

    daemon.restart().await;

    let listed = client.list_worktrees().await.expect("list after restart");
    assert_eq!(listed.len(), 1, "store must survive restart: {listed:?}");
    assert_eq!(listed[0].path, wt);
    assert_eq!(
        listed[0].reason,
        maestro_protocol::RetainedReason::ForceKilled
    );
    assert!(wt.is_dir(), "the tree itself must survive the restart too");

    client
        .reclaim_worktree(listed[0].id)
        .await
        .expect("reclaim");
    assert!(!wt.exists());
}

// ---------------------------------------------------------------------------
// GET /api/projects + /api/branches (#54/#55/#66)
// ---------------------------------------------------------------------------

/// #66: two repos under a configured `[spawn].project_roots` root are both
/// discovered by `GET /api/projects`; git activity gives every repo a
/// `last_activity`, and spawning inside one merges its command into the
/// project entry via spawn history.
#[tokio::test(flavor = "multi_thread")]
async fn projects_endpoint_discovers_roots_and_merges_activity() {
    let root = TempDir::new().unwrap();
    let alpha = make_repo(root.path(), "alpha");
    make_repo(root.path(), "beta");

    let daemon = Daemon::spawn_with_config(&format!(
        "[spawn]\nproject_roots = [\"{}\"]\nproject_scan_depth = 2\n",
        root.path().display()
    ))
    .await;
    let client = TestClient::new(&daemon.http_base());

    let projects = client.list_projects().await.expect("list projects");
    let names: Vec<&str> = projects.iter().map(|p| p.name.as_str()).collect();
    assert!(
        names.contains(&"alpha") && names.contains(&"beta"),
        "both repos must be discovered, got: {names:?}"
    );

    let find = |name: &str| {
        projects
            .iter()
            .find(|p| p.name == name)
            .cloned()
            .expect("project present")
    };
    // Both have commits, so git activity alone populates last_activity…
    assert!(
        find("alpha").last_activity.is_some(),
        "git activity must surface"
    );
    assert!(
        find("beta").last_activity.is_some(),
        "git activity must surface"
    );
    // …and neither has been spawned into yet.
    assert_eq!(find("alpha").last_command, None);
    assert_eq!(find("beta").last_command, None);

    // Spawning inside alpha writes durable history; /api/projects folds it in.
    let s = client
        .spawn_session(&create_req(&alpha, "sleep 30"))
        .await
        .expect("spawn in alpha");

    wait_for_async(
        "alpha's project entry reflects spawn history",
        Duration::from_secs(5),
        || {
            let client = client.clone();
            async move {
                client
                    .list_projects()
                    .await
                    .ok()
                    .map(|ps| {
                        ps.iter()
                            .find(|p| p.name == "alpha")
                            .map(|p| p.last_command.as_deref() == Some("sleep 30"))
                            .unwrap_or(false)
                    })
                    .unwrap_or(false)
            }
        },
    )
    .await;

    // beta was never spawned into: no command recorded there.
    let beta_after = client
        .list_projects()
        .await
        .expect("list after spawn")
        .into_iter()
        .find(|p| p.name == "beta")
        .unwrap();
    assert_eq!(beta_after.last_command, None);

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

/// #66: `GET /api/branches` lists every local branch, flags the main
/// checkout's HEAD with `is_head`, and reports which linked worktree holds an
/// occupied branch.
#[tokio::test(flavor = "multi_thread")]
async fn branches_endpoint_lists_branches_and_flags_occupied() {
    let (_daemon, client) = bring_up().await;

    let base = TempDir::new().unwrap();
    init_repo(base.path());
    run_git(&["branch", "feature/taken"], base.path());
    run_git(&["branch", "feature/free"], base.path());
    let holder = base.path().join("holder-wt");
    run_git(
        &[
            "worktree",
            "add",
            "-q",
            &holder.to_string_lossy(),
            "feature/taken",
        ],
        base.path(),
    );

    let branches = client
        .list_branches(&base.path().to_string_lossy())
        .await
        .expect("list branches");
    let by_name = |name: &str| {
        branches
            .iter()
            .find(|b| b.name == name)
            .cloned()
            .unwrap_or_else(|| panic!("branch {name} missing from {branches:?}"))
    };

    let main = by_name("main");
    assert!(main.is_head, "main is the main checkout's HEAD");
    assert_eq!(main.worktree, None);

    let taken = by_name("feature/taken");
    assert!(!taken.is_head);
    assert_eq!(
        taken.worktree.as_deref(),
        Some(holder.canonicalize().unwrap().as_path()),
        "occupied branch must report its holding worktree"
    );

    let free = by_name("feature/free");
    assert!(!free.is_head);
    assert_eq!(free.worktree, None, "unoccupied branch must be spawnable");
}

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

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

    // 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 after restart",
        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 (_daemon, client) = bring_up().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(&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 (daemon, client) = bring_up().await;

    // 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(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 client = daemon.client();
            let want = cwd.clone();
            async move {
                let Ok(recents) = client.recent_dirs().await else {
                    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 = daemon.client().git_root(&cwd).await.expect("gitroot");
    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 (_daemon, client) = bring_up().await;
    let cwd = TempDir::new().unwrap();

    let cat = client
        .spawn_session(&create_req(cwd.path(), "cat"))
        .await
        .expect("spawn cat");
    let slp = client
        .spawn_session(&create_req(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);
}

/// Real git repo at `<root>/<name>`, canonicalized (the scanner canonicalizes
/// discovered paths too, so comparisons must agree on the real path).
fn make_repo(root: &std::path::Path, name: &str) -> std::path::PathBuf {
    let dir = root.join(name);
    std::fs::create_dir_all(&dir).unwrap();
    init_repo(&dir);
    dir.canonicalize().unwrap()
}

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()
}

/// Bare remote registered as `origin` on `repo`, seeded with every current
/// branch, so existing commits are reachable from remote-tracking refs.
/// Without one the kill guard (#79) reads every commit as unpushed, and any
/// worktree session refuses to die. The caller must keep the `TempDir` alive
/// while the remote is still needed; later pushes go through [`push_all`].
fn throwaway_origin(repo: &std::path::Path) -> TempDir {
    let origin = tempfile::TempDir::new().unwrap();
    run_git(
        &["init", "-q", "--bare", origin.path().to_str().unwrap()],
        repo,
    );
    let _ = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["remote", "add", "origin", origin.path().to_str().unwrap()])
        .output()
        .expect("git remote add");
    push_all(repo);
    origin
}

fn push_all(repo: &std::path::Path) {
    run_git(&["push", "-q", "--all", "origin"], repo);
}

fn commit_all(dir: &std::path::Path, message: &str) {
    run_git(&["add", "-A"], dir);
    run_git(&["commit", "-q", "-m", message], dir);
}

fn branch_sha(repo: &std::path::Path, branch: &str) -> String {
    let out = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["rev-parse", "--verify", &format!("refs/heads/{branch}")])
        .output()
        .expect("git rev-parse");
    assert!(
        out.status.success(),
        "branch {branch} missing: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    String::from_utf8_lossy(&out.stdout).trim().to_string()
}

#[tokio::test(flavor = "multi_thread")]
async fn daemon_events_ws_streams_snapshot_then_session_added() {
    // 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 (daemon, client) = bring_up().await;
    let cwd = TempDir::new().unwrap();

    let ws_url = daemon.client().events_url().unwrap();
    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::CatalogEvent::Snapshot { sessions } => {
            assert!(sessions.is_empty());
        }
        other => panic!("expected initial Snapshot, got {other:?}"),
    }

    let session = client
        .spawn_session(&create_req(cwd.path(), "sleep 30"))
        .await
        .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::CatalogEvent::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");
}

// ---------------------------------------------------------------------------
// CLI against the local daemon
// ---------------------------------------------------------------------------

/// 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_daemon() {
    // The subject is the shipped `maestro` binary — arg parsing, config
    // precedence, output. It needs a daemon to talk to, not tmux, so this runs
    // against an in-memory one.
    let mock = maestro_testing::MockDaemon::start().await;
    let id = uuid::Uuid::new_v4();
    mock.add_session(maestro_testing::session(id, "/home/mock/work", "sleep 30"));

    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_DAEMON_URL", &mock.http_base()),
        ],
    );
    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(&id.to_string()),
        "session not listed: {stdout}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn cli_list_fails_when_daemon_is_down() {
    let dead_daemon = format!("http://127.0.0.1:{}", maestro_e2e::util::free_port());
    let home = TempDir::new().unwrap();
    let cfg_path = home.path().join("config.toml");
    let out = run_cli(
        &["list"],
        &[
            ("HOME", home.path().to_str().unwrap()),
            ("MAESTRO_CONFIG", cfg_path.to_str().unwrap()),
            ("MAESTRO_DAEMON_URL", &dead_daemon),
        ],
    );
    assert!(!out.status.success(), "expected a failure with no daemon");
}

async fn next_event<S>(ws: &mut S) -> maestro_protocol::CatalogEvent
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("CatalogEvent"),
            Some(Ok(_)) => continue,
            other => panic!("WS closed waiting for event: {other:?}"),
        }
    }
}