//! End-to-end tests for the SessionManager: create, kill, recover, broadcast.

mod common;

use std::collections::HashMap;
use std::path::PathBuf;
use std::process::Command;
use std::sync::Arc;
use std::time::Duration;

use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
use maestro_daemon::SessionError;
use maestro_protocol::CreateSessionRequest;
use maestro_protocol::{CatalogEvent, SessionStatus, WorktreeSpec};
use tempfile::TempDir;

#[tokio::test]
async fn create_then_kill_round_trip() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let session = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    assert!(fx.manager.has(session.id).await);
    let snap = fx.manager.snapshot().await;
    assert_eq!(snap.len(), 1);
    assert_eq!(snap[0].id, session.id);

    // sidecar should be on disk
    assert!(fx
        .sidecar_dir()
        .join(format!("{}.json", session.id))
        .exists());
    // FIFO should also exist
    assert!(fx.fifo_dir().join(format!("{}.fifo", session.id)).exists());

    fx.manager.kill(session.id, false).await.expect("kill");
    assert!(!fx.manager.has(session.id).await);
    assert!(!fx
        .sidecar_dir()
        .join(format!("{}.json", session.id))
        .exists());
}

// #24: a successful spawn appends a record to the durable history, which
// `recent_dirs()` then surfaces — proving the create→append→read path end to
// end against a real tmux + on-disk JSONL log.
#[tokio::test]
async fn create_appends_to_spawn_history() {
    let fx = DaemonFixture::new_with_history(30);
    let cwd = TempDir::new().unwrap();

    let session = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::from([("name".into(), serde_json::json!("neon-drifter"))]),
            worktree: None,
        })
        .await
        .expect("create");

    // The JSONL log exists and the recents query surfaces the chosen dir.
    assert!(fx.spawn_history_path().exists());
    let recents = fx.manager.recent_dirs();
    assert_eq!(recents.len(), 1);
    assert_eq!(recents[0].cwd, cwd.path());
    assert_eq!(recents[0].last_command, "sleep 30");

    fx.manager.kill(session.id, false).await.expect("kill");
    // History is durable — it survives the kill that deletes the sidecar.
    assert_eq!(fx.manager.recent_dirs().len(), 1);
}

// With history disabled (`0` days) or unwired, a spawn logs nothing and the
// recents query is empty — the spawn itself still succeeds.
#[tokio::test]
async fn create_without_history_logs_nothing() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    fx.manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");
    assert!(fx.manager.recent_dirs().is_empty());
    assert!(!fx.spawn_history_path().exists());
}

#[tokio::test]
async fn recover_adopts_existing_tmux_sessions() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    // create a session, then "throw away" the SessionManager (the tmux session
    // and sidecar persist) and build a fresh manager that recovers them.
    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    // build a parallel manager pointing at the same dirs + same tmux server
    let sidecar = maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir());
    let worktree_mgr = maestro_daemon::worktree::WorktreeManager::new(None);
    let revived = maestro_daemon::SessionManager::new(
        Arc::new(fx.tmux.tmux.clone()),
        sidecar,
        worktree_mgr,
        fx.fifo_dir(),
    );
    let adopted = revived.recover().await.expect("recover");
    assert_eq!(adopted, 1);
    assert!(revived.has(s.id).await);
    let snap = revived.snapshot().await;
    assert_eq!(snap[0].id, s.id);
    assert_eq!(snap[0].command, "sleep 30");
}

/// Regression for NAP-253: the very first bytes a freshly-spawned command
/// writes must reach the replay buffer. Before the fix, `pipe-pane` was a
/// separate tmux invocation that ran after `new-session` + an async FIFO
/// setup, so the child's first output was buffered by tmux but never tee'd
/// into the FIFO.
#[tokio::test]
async fn create_captures_initial_output_in_replay_buffer() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            // The shell prints the marker before doing anything else and
            // then sleeps so the session stays alive. Before NAP-253 this
            // marker was the canonical case that got swallowed.
            command: "sh -c 'echo NAP_253_FIRST_LINE_MARKER; exec sleep 30'".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    let want = b"NAP_253_FIRST_LINE_MARKER";
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    let mut found = false;
    while std::time::Instant::now() < deadline {
        let (replay, _rx) = fx
            .manager
            .subscribe_pty_with_replay(s.id)
            .await
            .expect("subscribe");
        if replay.windows(want.len()).any(|w| w == want) {
            found = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(
        found,
        "first-line marker never reached the replay buffer — pipe-pane raced past the child's initial output",
    );

    fx.manager.kill(s.id, false).await.unwrap();
}

/// A fresh attach must land on the pane's *current* screen. Replaying the raw
/// byte tail instead reproduced a browser refresh painting garbage over an
/// agent TUI: the tail starts mid-stream, so the clears, modes and cursor
/// moves that gave those bytes meaning are long gone.
#[tokio::test]
async fn attach_replay_is_the_current_screen_not_the_byte_tail() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            // Print a line, then switch to the alternate screen the way a
            // full-screen program does on startup. Only the second marker is
            // on the pane; the first belongs to the screen behind it.
            command: "sh -c 'printf STALE_MARKER_LINE\\\\n; sleep 0.5; printf \"\\\\033[?1049h\\\\033[2J\\\\033[HLIVE_MARKER\"; exec sleep 30'"
                .into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    let want = b"LIVE_MARKER";
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    let mut replay = Vec::new();
    while std::time::Instant::now() < deadline {
        let (bytes, _rx) = fx
            .manager
            .subscribe_pty_with_replay(s.id)
            .await
            .expect("subscribe");
        replay = bytes;
        if replay.windows(want.len()).any(|w| w == want) {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    let text = String::from_utf8_lossy(&replay).into_owned();
    assert!(
        replay.starts_with(b"\x1b[?1049h\x1b[H\x1b[2J"),
        "replay did not start with an alternate-screen snapshot: {text:?}"
    );
    assert!(
        text.contains("LIVE_MARKER"),
        "replay lost the live screen: {text:?}"
    );
    assert!(
        !text.contains("STALE_MARKER_LINE"),
        "replay carried bytes from behind the alternate screen: {text:?}"
    );

    fx.manager.kill(s.id, false).await.unwrap();
}

/// Regression for NAP-228: after `recover`, PTY output must reach new
/// subscribers. Before the fix tmux's stale `pipe-pane` (with `-o`) silently
/// kept feeding the old FIFO and the revived manager's subscribers saw
/// nothing.
#[tokio::test]
async fn recover_re_pipes_pane_so_output_reaches_subscribers() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "bash --noprofile --norc".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    // Capture paths + tmux handle before dropping the original manager —
    // `drop(fx.manager)` partially moves `fx` so we can't call its helpers
    // afterwards.
    let sidecar_dir = fx.sidecar_dir();
    let fifo_dir = fx.fifo_dir();
    let tmux = fx.tmux.tmux.clone();

    // Drop the original manager. Its PtyBridge closes the FIFO's only
    // reader, so tmux's `cat >> fifo` writer hits SIGPIPE and exits — but
    // tmux's pipe-pane bookkeeping still thinks the pane is piped. That's
    // exactly the stale state recover() has to overwrite.
    drop(fx.manager);

    let sidecar = maestro_daemon::sidecar::SidecarStore::new(sidecar_dir);
    let worktree_mgr = maestro_daemon::worktree::WorktreeManager::new(None);
    let revived =
        maestro_daemon::SessionManager::new(Arc::new(tmux), sidecar, worktree_mgr, fifo_dir);
    assert_eq!(revived.recover().await.expect("recover"), 1);

    let mut rx = revived.subscribe_pty(s.id).await.expect("subscribe");
    revived
        .send_keys(s.id, b"echo NAP_228_RECOVERED\n")
        .await
        .unwrap();

    let want = b"NAP_228_RECOVERED";
    let mut buf: Vec<u8> = Vec::new();
    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    while std::time::Instant::now() < deadline {
        if let Ok(Ok(bytes)) = tokio::time::timeout(Duration::from_millis(50), rx.recv()).await {
            buf.extend_from_slice(&bytes);
            if buf.windows(want.len()).any(|w| w == want) {
                break;
            }
        }
    }
    assert!(
        buf.windows(want.len()).any(|w| w == want),
        "recover() did not restore pty output flow; buf={:?}",
        String::from_utf8_lossy(&buf),
    );

    revived.kill(s.id, false).await.unwrap();
}

#[tokio::test]
async fn pty_broadcast_delivers_bytes_to_subscribers() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "bash --noprofile --norc".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .unwrap();

    let mut a = fx.manager.subscribe_pty(s.id).await.unwrap();
    let mut b = fx.manager.subscribe_pty(s.id).await.unwrap();

    fx.manager
        .send_keys(s.id, b"echo MAESTRO_BROADCAST_OK\n")
        .await
        .unwrap();

    let want = b"MAESTRO_BROADCAST_OK";
    let mut buf_a: Vec<u8> = Vec::new();
    let mut buf_b: Vec<u8> = Vec::new();

    let deadline = std::time::Instant::now() + Duration::from_secs(5);
    while std::time::Instant::now() < deadline {
        if let Ok(Ok(bytes)) = tokio::time::timeout(Duration::from_millis(50), a.recv()).await {
            buf_a.extend_from_slice(&bytes);
        }
        if let Ok(Ok(bytes)) = tokio::time::timeout(Duration::from_millis(50), b.recv()).await {
            buf_b.extend_from_slice(&bytes);
        }
        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), "A missed");
    assert!(buf_b.windows(want.len()).any(|w| w == want), "B missed");

    fx.manager.kill(s.id, false).await.unwrap();
    let _ = (cwd, PathBuf::from("/"));
}

#[tokio::test]
async fn liveness_watcher_marks_dead_sessions_and_keeps_them_until_dismissed() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    // Subscribe *after* the create so we don't have to filter the Added event.
    let mut events = fx.manager.events();
    fx.manager.start_liveness_watcher(Duration::from_millis(50));

    // Kill the tmux session out from under the daemon. This is the scenario
    // the watcher exists for: command exits, tmux destroys the session, the
    // daemon never finds out through its normal channels.
    fx.tmux.tmux.kill_session(s.id).expect("kill tmux directly");

    wait_for_async(Duration::from_secs(3), || async {
        fx.manager.status(s.id).await == Some(SessionStatus::Dead)
    })
    .await;

    // Marked, not removed — the operator still sees it exited, and its
    // artifacts stay so the output can be replayed.
    assert!(fx.manager.has(s.id).await);
    assert!(fx.sidecar_dir().join(format!("{}.json", s.id)).exists());

    let mut saw_dead = false;
    let deadline = std::time::Instant::now() + Duration::from_secs(1);
    while std::time::Instant::now() < deadline && !saw_dead {
        match tokio::time::timeout(Duration::from_millis(50), events.recv()).await {
            Ok(Ok(CatalogEvent::SessionUpdated { session }))
                if session.id == s.id && session.status == SessionStatus::Dead =>
            {
                saw_dead = true
            }
            Ok(Ok(_)) => continue,
            _ => break,
        }
    }
    assert!(
        saw_dead,
        "expected a SessionUpdated marking the session dead"
    );

    // Dismissing it is what finally releases the artifacts.
    fx.manager.kill(s.id, false).await.expect("dismiss");
    assert!(!fx.manager.has(s.id).await);
    assert!(!fx.sidecar_dir().join(format!("{}.json", s.id)).exists());
    assert!(!fx.fifo_dir().join(format!("{}.fifo", s.id)).exists());
}

/// #83: the watcher refreshes `Session.blocked_work` on its own cadence —
/// first "nothing blocks" for a pristine tree, then the breakdown once the
/// tree picks up unsaved work, so clients can warn before the kill attempt.
#[tokio::test]
async fn watcher_refreshes_the_blocked_work_indicator() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("indicator");

    fx.manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/indicator".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");

    // Pushed and clean: nothing in the way, and the indicator must say so.
    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());

    // Subscribe before the watcher starts so every pass's emission is
    // captured — otherwise the breakdown event can land between the
    // snapshot observation below and the drain, and the quiet window
    // would see it as an illegitimate re-emit.
    let mut events = fx.manager.events();

    fx.manager.start_liveness_watcher_with_blocked_work(
        Duration::from_millis(50),
        Duration::from_millis(100),
    );

    wait_for_async(Duration::from_secs(5), || async {
        fx.manager.snapshot().await[0].blocked_work == Some(maestro_protocol::InTheWay::default())
    })
    .await;

    // Unsaved work appears; the next pass must surface it.
    std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();
    std::fs::write(wt_path.join("notes.txt"), b"untracked\n").unwrap();
    wait_for_async(Duration::from_secs(5), || async {
        matches!(
            &fx.manager.snapshot().await[0].blocked_work,
            Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1
        )
    })
    .await;

    // Consume through the emission that carried the breakdown (#83
    // change-detection): the quiet window below must start after the last
    // legitimate change, not race it into the channel.
    loop {
        match tokio::time::timeout(Duration::from_secs(5), events.recv()).await {
            Ok(Ok(CatalogEvent::SessionUpdated { session }))
                if matches!(
                    &session.blocked_work,
                    Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1
                ) =>
            {
                break;
            }
            Ok(Ok(_)) => continue,
            _ => panic!("never saw the SessionUpdated carrying the breakdown"),
        }
    }
    while events.try_recv().is_ok() {}
    tokio::time::sleep(Duration::from_millis(400)).await;
    // Only a *changed* indicator counts as a re-emit (#83 change-detection).
    // The activity pass may still legitimately announce e.g. Active→Idle
    // (its debounce is time-based, not worktree-based), and its payload
    // carries today's blocked_work unchanged.
    while let Ok(CatalogEvent::SessionUpdated { session }) = events.try_recv() {
        assert!(
            matches!(
                &session.blocked_work,
                Some(b) if b.uncommitted_files >= 1 && b.untracked_files >= 1
            ),
            "an unchanged worktree must not re-emit a changed indicator"
        );
    }
}

/// Cleanup on kill is unconditional once the guard passes (#79): even a spec
/// that says "don't clean up" loses its worktree — the branch was pushed, so
/// nothing in the tree is unique and removal goes ahead.
#[tokio::test]
async fn kill_removes_worktree_even_when_spec_says_keep() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("keep");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/keep".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");
    assert!(wt_path.exists());

    // Push the branch so its commits are reachable from a remote-tracking ref.
    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());

    fx.manager.kill(s.id, false).await.expect("kill");
    assert!(!wt_path.exists(), "worktree must be removed on kill");
}

/// The kill guard (#79): a worktree holding uncommitted changes refuses the
/// kill outright. Nothing happens — the session keeps running as the
/// operator's handle on their unfinished work, and the tree is untouched.
/// The refusal names what blocked it and how much of it there is.
#[tokio::test]
async fn killing_refuses_a_worktree_with_unsaved_work_and_keeps_the_session() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("guarded");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/guarded".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");
    std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();
    std::fs::write(wt_path.join("notes.txt"), b"untracked\n").unwrap();

    let err = fx
        .manager
        .kill(s.id, false)
        .await
        .expect_err("kill must refuse");
    let SessionError::KillGuarded(breakdown) = err else {
        panic!("expected KillGuarded, got {err:?}");
    };
    assert_eq!(breakdown.uncommitted_files, 1);
    assert_eq!(breakdown.untracked_files, 1);
    assert!(
        breakdown.unpushed_commits >= 1,
        "a fresh branch has no remote-tracking ref"
    );
    let message = breakdown.to_string();
    assert!(message.contains("1 uncommitted file"), "{message}");
    assert!(message.contains("1 untracked file"), "{message}");

    // Nothing happened: the session survives and so does the tree.
    assert!(fx.manager.has(s.id).await, "the session must keep running");
    assert!(wt_path.exists(), "the worktree must be untouched");

    // A second kill after the work is saved goes through.
    run_git_in(&wt_path, &["add", "."]);
    run_git_in(&wt_path, &["commit", "-q", "-m", "saved"]);
    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());
    fx.manager
        .kill(s.id, false)
        .await
        .expect("kill after saving");
    assert!(!fx.manager.has(s.id).await);
}

/// Unpushed commits alone block the kill (#79): a feature branch almost
/// always has them, which is why `--force` (which retains) is the ordinary
/// way to end such a session.
#[tokio::test]
async fn killing_refuses_until_the_branch_is_pushed() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("pushed");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/pushed".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");

    let err = fx
        .manager
        .kill(s.id, false)
        .await
        .expect_err("unpushed blocks");
    let SessionError::KillGuarded(breakdown) = err else {
        panic!("expected KillGuarded, got {err:?}");
    };
    assert_eq!(breakdown.uncommitted_files, 0);
    assert_eq!(breakdown.untracked_files, 0);
    assert!(breakdown.unpushed_commits >= 1);

    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());

    fx.manager
        .kill(s.id, false)
        .await
        .expect("pushed branch kills");
    assert!(!wt_path.exists());
    assert!(!fx.manager.has(s.id).await);
}

/// `--force` overrides the refusal and nothing else (#81): the session ends,
/// but the guarded worktree is left on disk — never removed.
#[tokio::test]
async fn force_killing_a_guarded_worktree_leaves_it_on_disk() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("forced");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/forced".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");
    std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();

    // Without force the same kill is refused (#79).
    assert!(fx.manager.kill(s.id, false).await.is_err());

    fx.manager
        .kill(s.id, true)
        .await
        .expect("force overrides the refusal");

    // The tree would have blocked, so force must NOT have removed it...
    assert!(
        wt_path.exists(),
        "a forced kill never removes a tree that would have blocked"
    );
    // ...but the session itself still ends.
    assert!(!fx.manager.has(s.id).await);
    assert!(!fx.sidecar_dir().join(format!("{}.json", s.id)).exists());
}

/// Force overrides the *refusal*, not ordinary cleanup (#81): a tree whose
/// guard is clear never needed the override, so a forced kill removes it as
/// usual.
#[tokio::test]
async fn force_on_a_clear_worktree_removes_it_as_usual() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("clear");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/clear".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");

    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());

    fx.manager.kill(s.id, true).await.expect("kill");
    assert!(!wt_path.exists(), "a clear tree needs no override");
}

/// Ignored files NEVER block (#79): build output like `target/` exists in
/// every Rust checkout, and an ignored-only worktree stays killable.
#[tokio::test]
async fn killing_an_ignored_only_worktree_succeeds() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    // .gitignore must be committed before the worktree branches off, or it
    // would itself read as an untracked file.
    std::fs::write(repo.path().join(".gitignore"), b"/target\n").unwrap();
    run_git_in(repo.path(), &["add", ".gitignore"]);
    run_git_in(repo.path(), &["commit", "-q", "-m", "ignore target"]);
    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());

    let wt_path = repo.path().join("worktrees").join("feat").join("rusty");
    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/rusty".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");
    std::fs::create_dir_all(wt_path.join("target")).unwrap();
    std::fs::write(wt_path.join("target").join("app.o"), b"binary\n").unwrap();

    // Branch pushed; the only dirt is ignored output.
    push_all_to_origin(repo.path(), origin.path());

    fx.manager
        .kill(s.id, false)
        .await
        .expect("ignored files never block");
    assert!(!wt_path.exists());
    assert!(!fx.manager.has(s.id).await);
}

/// Natural exit (#82): a session that ends on its own never reaches the kill
/// path, so the watcher applies the kill's rule itself. A worktree holding
/// unsaved work is LEFT ON DISK and stays attached to the dead record; the
/// session stays in the catalog marked dead until dismissed, and dismissing
/// it goes through `--force`, which leaves the tree alone too.
#[tokio::test]
async fn natural_exit_of_a_guarded_worktree_leaves_it_on_disk() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("exited");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/exited".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");
    std::fs::write(wt_path.join("README"), b"unsaved edits\n").unwrap();

    // The command finishes / the operator types exit: tmux loses the session.
    fx.tmux.tmux.kill_session(s.id).expect("kill tmux directly");
    fx.manager.start_liveness_watcher(Duration::from_millis(50));
    wait_for_async(Duration::from_secs(3), || async {
        fx.manager.status(s.id).await == Some(SessionStatus::Dead)
    })
    .await;

    // The settle step has been applied or refused either way — here refused,
    // since the tree holds unsaved work. Only from this point is "the tree
    // survived" a settled fact rather than a race (#115).
    wait_for_async(Duration::from_secs(3), || async {
        fx.manager.is_settled(s.id).await == Some(true)
    })
    .await;

    // The tree would have blocked a kill, so the exit kept it.
    assert!(
        wt_path.exists(),
        "a dirty worktree must be left alone on natural exit"
    );
    let snap = fx.manager.snapshot().await;
    assert_eq!(snap.len(), 1);
    assert!(
        snap[0].worktree.is_some(),
        "the tree stays attached to the dead record"
    );

    // Dismissing it needs --force, and force never removes what would block.
    assert!(fx.manager.kill(s.id, false).await.is_err());
    fx.manager.kill(s.id, true).await.expect("forced dismissal");
    assert!(!fx.manager.has(s.id).await);
    assert!(wt_path.exists(), "dismissal must not remove the work");
}

/// Natural exit (#82), pristine side of the same rule: nothing in the tree,
/// nothing to keep — the worktree is removed.
#[tokio::test]
async fn natural_exit_of_a_pristine_worktree_removes_it() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());
    let wt_path = repo.path().join("worktrees").join("feat").join("clean");

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/clean".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create");

    // Push everything so the branch has no unpushed commits: fully pristine.
    let origin = TempDir::new().unwrap();
    init_bare_repo(origin.path());
    push_all_to_origin(repo.path(), origin.path());

    fx.tmux.tmux.kill_session(s.id).expect("kill tmux directly");
    fx.manager.start_liveness_watcher(Duration::from_millis(50));
    wait_for_async(Duration::from_secs(3), || async {
        fx.manager.status(s.id).await == Some(SessionStatus::Dead)
    })
    .await;

    // (`Dead` is observable a moment before the settle step removes the tree.)
    wait_for_async(Duration::from_secs(3), || async { !wt_path.exists() }).await;

    assert!(
        !wt_path.exists(),
        "a pristine worktree leaves nothing behind"
    );
    assert!(
        fx.manager.has(s.id).await,
        "the session stays until dismissed"
    );
}

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

#[tokio::test]
async fn liveness_watcher_leaves_running_sessions_alone() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    fx.manager.start_liveness_watcher(Duration::from_millis(50));

    // Give the watcher enough time to run a handful of ticks.
    tokio::time::sleep(Duration::from_millis(400)).await;
    assert!(fx.manager.has(s.id).await, "live session got reaped");

    fx.manager.kill(s.id, false).await.unwrap();
}

/// #99: clients group a session under `worktree.base_repo`, so that field has
/// to name the repository even when the spawn request pointed at a linked
/// worktree. Recording the request verbatim put the session in a folder of its
/// own — `<repo>/worktrees/<branch>` — instead of under the repo.
#[tokio::test]
async fn worktree_base_repo_records_the_main_checkout_not_the_request() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    init_repo(repo.path());

    // A linked worktree of that repo, then a spawn whose `base_repo` names it
    // rather than the main checkout — what happens when the operator spawns
    // from inside a worktree.
    let first = fx
        .manager
        .create(CreateSessionRequest {
            cwd: repo.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: repo.path().to_path_buf(),
                branch: "feat/one".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create first");
    let inner_worktree = first.worktree.expect("first has a worktree").path;

    let second = fx
        .manager
        .create(CreateSessionRequest {
            cwd: inner_worktree.clone(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: Some(WorktreeSpec {
                base_repo: inner_worktree.clone(),
                branch: "feat/two".into(),
                create_branch: true,
            }),
        })
        .await
        .expect("create second");

    let wt = second.worktree.expect("second has a worktree");
    assert_eq!(
        wt.base_repo,
        repo.path()
            .canonicalize()
            .unwrap_or_else(|_| repo.path().to_path_buf()),
        "base_repo must be the repository, not the worktree the request named",
    );
    assert_ne!(wt.base_repo, inner_worktree, "the worktree is not the repo");
}

/// #103: every tmux-spawned session carries the daemon's identity variables in
/// its environment — its UUID under `MAESTRO_SESSION_ID` and the daemon's base
/// URL under `MAESTRO_URL` — so the in-session reporting hook (#106) can find
/// both without any out-of-band channel.
#[tokio::test]
async fn spawned_sessions_export_maestro_session_id_and_url() {
    let fx = DaemonFixture::new();
    fx.manager
        .set_daemon_url("http://127.0.0.1:7071".to_string());
    let cwd = TempDir::new().unwrap();

    let s = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: "sleep 30".into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create");

    let env = fx
        .tmux
        .tmux
        .show_environment(s.id)
        .expect("show-environment");
    assert_eq!(
        env.get("MAESTRO_SESSION_ID"),
        Some(&s.id.to_string()),
        "session env: {env:?}"
    );
    assert_eq!(
        env.get("MAESTRO_URL"),
        Some(&"http://127.0.0.1:7071".to_string())
    );

    fx.manager.kill(s.id, false).await.expect("kill");
}