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

//! The session manager's decisions, driven through [`FakePanes`].
//!
//! These are the branches that used to need a forked tmux server to reach:
//! spawn rollback ordering, liveness diffing, and activity change-detection.
//! Nothing here spawns a process, so they run everywhere and in milliseconds.

use std::collections::{BTreeMap, HashMap};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use maestro_daemon::panes::{Call, FakePanes, Op};
use maestro_daemon::session::{SessionError, SessionManager};
use maestro_daemon::sidecar::SidecarStore;
use maestro_daemon::tmux::PaneInfo;
use maestro_daemon::worktree::WorktreeManager;
use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionStatus};
use tempfile::TempDir;
use tokio::sync::broadcast::error::TryRecvError;
use uuid::Uuid;

struct Fx {
    manager: SessionManager,
    fake: Arc<FakePanes>,
    tmp: TempDir,
}

impl Fx {
    fn new() -> Self {
        Self::with_sidecar_dir(|tmp| tmp.join("sidecars"))
    }

    fn with_sidecar_dir(pick: impl Fn(&std::path::Path) -> PathBuf) -> Self {
        let tmp = TempDir::new().expect("tempdir");
        let fake = Arc::new(FakePanes::new());
        let fifo_dir = tmp.path().join("fifos");
        std::fs::create_dir_all(&fifo_dir).expect("fifo dir");
        let manager = SessionManager::new(
            fake.clone(),
            SidecarStore::new(pick(tmp.path())),
            WorktreeManager::new(tmp.path().join("worktrees")),
            fifo_dir,
        );
        Self { manager, fake, tmp }
    }

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

/// Poll until `id` reaches `want`.
async fn wait_for_status(manager: &SessionManager, id: uuid::Uuid, want: SessionStatus) {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline {
        if manager.status(id).await == Some(want) {
            return;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    panic!("timed out waiting for {id} to become {want:?}");
}

/// NAP-260 rollback, asserted as an *ordering* rather than an end state.
///
/// Replaces a test that forced the sidecar failure with `chmod 0o500` and then
/// returned early when it detected root — so on CI it asserted nothing. Here
/// the sidecar directory cannot be created because a regular file sits where
/// its parent must be, which no uid can work around.
#[tokio::test]
async fn create_kills_the_tmux_session_when_the_sidecar_write_fails() {
    let fx = Fx::with_sidecar_dir(|tmp| {
        let blocker = tmp.join("blocker");
        std::fs::write(&blocker, b"not a directory").expect("blocker file");
        blocker.join("sidecars")
    });

    let res = fx.manager.create(fx.request("sleep 30")).await;

    assert!(
        matches!(res, Err(SessionError::Sidecar(_))),
        "expected the sidecar write to fail, got {res:?}"
    );

    let calls = fx.fake.calls();
    assert_eq!(
        calls.len(),
        2,
        "expected spawn then rollback kill, got {calls:?}"
    );
    assert!(matches!(calls[0], Call::NewSessionPiped { .. }));
    let Call::NewSessionPiped { id, .. } = calls[0] else {
        unreachable!()
    };
    assert_eq!(
        calls[1],
        Call::KillSession(id),
        "the rollback must kill the session it just spawned"
    );

    assert!(!fx.fake.is_live(id), "create left a ghost tmux session");
    assert!(fx.manager.snapshot().await.is_empty());
}

/// A failed spawn must not roll back anything, least of all kill a session id
/// tmux never accepted.
#[tokio::test]
async fn create_does_not_roll_back_when_the_spawn_itself_fails() {
    let fx = Fx::new();
    fx.fake.fail(Op::NewSessionPiped, "no server running");

    let res = fx.manager.create(fx.request("sleep 30")).await;
    assert!(res.is_err(), "expected the spawn to fail, got {res:?}");

    assert_eq!(
        fx.fake.calls().len(),
        1,
        "only the failed spawn should have been attempted: {:?}",
        fx.fake.calls()
    );
    assert!(fx.manager.snapshot().await.is_empty());
}

/// A restarted daemon adopts whatever tmux still holds, and must re-pipe each
/// one — without that the adopted session produces no output.
#[tokio::test]
async fn recover_adopts_live_sessions_and_repipes_them() {
    let fx = Fx::new();
    let orphan = Uuid::new_v4();
    fx.fake.add_live(orphan);

    let adopted = fx.manager.recover().await.expect("recover");
    assert_eq!(adopted, 1);

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

    assert!(
        fx.fake.calls().contains(&Call::PipePane(orphan)),
        "adopted session was never re-piped: {:?}",
        fx.fake.calls()
    );
}

/// A session tmux lost is marked `Dead` and kept, so the operator can still see
/// it exited and read what it printed. It is not removed.
#[tokio::test]
async fn liveness_marks_the_lost_session_dead_and_keeps_it() {
    let fx = Fx::new();
    let keep = fx.manager.create(fx.request("keep")).await.expect("keep");
    let lose = fx.manager.create(fx.request("lose")).await.expect("lose");
    let mut events = fx.manager.events();

    fx.fake.remove_live(lose.id);
    fx.manager.start_liveness_watcher(Duration::from_millis(25));
    wait_for_status(&fx.manager, lose.id, SessionStatus::Dead).await;

    let all = fx.manager.snapshot().await;
    assert_eq!(all.len(), 2, "the dead session must stay in the catalog");
    let survivor = all.iter().find(|s| s.id == keep.id).expect("keep");
    assert_eq!(survivor.status, SessionStatus::Running);

    let mut removed = Vec::new();
    while let Ok(ev) = events.try_recv() {
        if let CatalogEvent::SessionRemoved { id } = ev {
            removed.push(id);
        }
    }
    assert!(removed.is_empty(), "nothing should be removed: {removed:?}");
}

/// The transition fires once. Without the Running filter the pass re-announces
/// the same dead session on every tick, forever.
#[tokio::test]
async fn a_dead_session_is_announced_only_once() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");
    let mut events = fx.manager.events();

    fx.fake.remove_live(s.id);
    fx.manager.start_liveness_watcher(Duration::from_millis(25));
    wait_for_status(&fx.manager, s.id, SessionStatus::Dead).await;

    let mut dead_updates = 0;
    let deadline = tokio::time::Instant::now() + Duration::from_millis(300);
    while tokio::time::Instant::now() < deadline {
        if let Ok(CatalogEvent::SessionUpdated { session }) = events.try_recv() {
            if session.status == SessionStatus::Dead {
                dead_updates += 1;
            }
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert_eq!(dead_updates, 1, "expected exactly one Dead announcement");
}

/// Dismissing a dead session releases it — without asking tmux to kill
/// something tmux already lost.
#[tokio::test]
async fn killing_a_dead_session_dismisses_it_without_touching_tmux() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");
    fx.fake.remove_live(s.id);
    fx.manager.start_liveness_watcher(Duration::from_millis(25));
    wait_for_status(&fx.manager, s.id, SessionStatus::Dead).await;

    fx.manager.kill(s.id).await.expect("dismiss");

    assert!(fx.manager.snapshot().await.is_empty());
    assert!(
        !fx.fake
            .calls()
            .iter()
            .any(|c| matches!(c, Call::KillSession(k) if *k == s.id)),
        "kill-session was called on a session tmux had already lost: {:?}",
        fx.fake.calls()
    );
}

/// A dead session has no pane, so input is dropped rather than surfacing as a
/// tmux failure the client would have to interpret.
#[tokio::test]
async fn a_dead_session_is_read_only() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");
    fx.fake.remove_live(s.id);
    fx.manager.start_liveness_watcher(Duration::from_millis(25));
    wait_for_status(&fx.manager, s.id, SessionStatus::Dead).await;

    fx.manager
        .send_keys(s.id, b"ls\n")
        .await
        .expect("send_keys");
    fx.manager.resize(s.id, 80, 24).await.expect("resize");

    assert!(
        !fx.fake
            .calls()
            .iter()
            .any(|c| matches!(c, Call::SendKeys { .. } | Call::ResizeWindow { .. })),
        "input reached tmux for a dead session: {:?}",
        fx.fake.calls()
    );
}

/// `CreateSessionRequest.env` was accepted, stored, echoed back to clients, and
/// then dropped: `new_session_piped` had no env parameter at all.
#[tokio::test]
async fn requested_env_reaches_tmux() {
    let fx = Fx::new();
    let mut req = fx.request("bash");
    req.env.insert("FOO".into(), "bar".into());
    req.env.insert("EMPTY".into(), String::new());

    fx.manager.create(req).await.expect("create");

    let want: BTreeMap<String, String> = [
        ("EMPTY".to_string(), String::new()),
        ("FOO".to_string(), "bar".to_string()),
    ]
    .into_iter()
    .collect();
    match fx.fake.calls().first() {
        Some(Call::NewSessionPiped { env, .. }) => assert_eq!(env, &want),
        other => panic!("expected a spawn call, got {other:?}"),
    }
}

#[tokio::test]
async fn an_env_key_tmux_would_misparse_is_rejected() {
    let fx = Fx::new();
    let mut req = fx.request("bash");
    req.env.insert("A=B".into(), "x".into());

    let err = fx.manager.create(req).await.expect_err("should reject");
    assert_eq!(err.to_error_info().code, "invalid_env");
    assert!(
        fx.fake.calls().is_empty(),
        "nothing should have been spawned: {:?}",
        fx.fake.calls()
    );
}

/// If `list_sessions` fails the pass has no idea what is alive. Reaping on that
/// basis would kill the whole catalog.
#[tokio::test]
async fn liveness_reaps_nothing_when_tmux_cannot_be_listed() {
    let fx = Fx::new();
    let alive = fx.manager.create(fx.request("bash")).await.expect("create");

    fx.fake.fail(Op::ListSessions, "connection refused");
    fx.manager.start_liveness_watcher(Duration::from_millis(25));

    // Give the watcher several passes to do the wrong thing.
    tokio::time::sleep(Duration::from_millis(200)).await;

    let snap = fx.manager.snapshot().await;
    assert_eq!(
        snap.len(),
        1,
        "a failed list must not be read as 'everything died'"
    );
    assert_eq!(snap[0].id, alive.id);
}

#[tokio::test]
async fn activity_pass_emits_updated_only_when_something_changed() {
    let fx = Fx::new();
    let session = fx.manager.create(fx.request("zsh")).await.expect("create");
    let mut events = fx.manager.events();

    fx.fake.set_pane_info(
        session.id,
        PaneInfo {
            // A pid nothing can be reading from, so `awaiting_input` is a
            // deterministic false rather than a probe of this machine.
            pane_pid: i32::MAX,
            command: "claude".into(),
            tty: "/dev/null".into(),
            current_path: "/srv/work".into(),
        },
    );
    fx.manager.start_liveness_watcher(Duration::from_millis(25));

    let updated = wait_for_updated(&mut events).await;
    assert_eq!(updated.foreground.as_deref(), Some("claude"));
    assert_eq!(
        updated.current_cwd.as_deref(),
        Some(std::path::Path::new("/srv/work"))
    );
    // Still inside `activity::ACTIVE_WINDOW` (3s) since the spawn, so the
    // change here is `foreground`, not the activity state.
    assert_eq!(updated.activity, ActivityState::Active);

    // Drain anything already queued, then confirm a steady pane stays quiet.
    // Stays well inside the 3s window, so no Active -> Idle transition either.
    while events.try_recv().is_ok() {}
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(
        matches!(events.try_recv(), Err(TryRecvError::Empty)),
        "an unchanged pane must not re-emit SessionUpdated every tick"
    );
}

async fn wait_for_updated(
    events: &mut tokio::sync::broadcast::Receiver<CatalogEvent>,
) -> maestro_protocol::Session {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline {
        match events.try_recv() {
            Ok(CatalogEvent::SessionUpdated { session }) => return session,
            Ok(_) => continue,
            Err(_) => tokio::time::sleep(Duration::from_millis(10)).await,
        }
    }
    panic!("timed out waiting for SessionUpdated");
}