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

    /// Variant whose manager knows the daemon's own base URL (#103), the way
    /// the production runner wires it from `[daemon].bind_addr`.
    fn with_daemon_url(url: &str) -> Self {
        let fx = Self::new();
        fx.manager.set_daemon_url(url.to_string());
        fx
    }

    /// The env map of the (single, expected) spawn call.
    fn spawned_env(&self) -> BTreeMap<String, String> {
        let calls = self.fake.calls();
        let Some(Call::NewSessionPiped { env, .. }) = calls.first() else {
            panic!("expected a spawn call, got {calls:?}");
        };
        assert_eq!(calls.len(), 1, "expected exactly one spawn: {calls:?}");
        env.clone()
    }

    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(Some(tmp.path().join("worktrees"))),
            fifo_dir,
        );
        Self { manager, fake, tmp }
    }

    fn request(&self, command: &str) -> CreateSessionRequest {
        CreateSessionRequest {
            view: Default::default(),
            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, false).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 spawned = match fx.fake.calls().first() {
        Some(Call::NewSessionPiped { env, .. }) => env.clone(),
        other => panic!("expected a spawn call, got {other:?}"),
    };
    let id = spawned
        .get("MAESTRO_SESSION_ID")
        .expect("spawn must export MAESTRO_SESSION_ID (#103)")
        .clone();
    // The daemon-owned variable rides alongside whatever was requested.
    let want: BTreeMap<String, String> = [
        ("EMPTY".to_string(), String::new()),
        ("FOO".to_string(), "bar".to_string()),
        ("MAESTRO_SESSION_ID".to_string(), id),
    ]
    .into_iter()
    .collect();
    assert_eq!(&spawned, &want);
}

#[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);
    // A foreground/cwd-only change is not a state transition (#104): the
    // sidebar's duration must keep counting from the original stamp.
    assert_eq!(updated.activity_since, session.activity_since);

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

/// The sidebar's duration (#104) is time in the *current* state, so
/// `activity_since` is stamped exactly when the decided state differs from
/// the stored one — once per transition. Stamping it every tick would make
/// the duration permanently read zero.
#[tokio::test]
async fn activity_transition_stamps_activity_since_once_per_transition() {
    let fx = Fx::new();
    let session = fx
        .manager
        .create(fx.request("sleep 300"))
        .await
        .expect("create");
    fx.fake.set_pane_info(
        session.id,
        PaneInfo {
            // Nothing reads from this pid: `awaiting_input` is deterministically
            // false, so past `ACTIVE_WINDOW` the session decides as Idle.
            pane_pid: i32::MAX,
            command: "sleep".into(),
            tty: "/dev/null".into(),
            current_path: String::new(),
        },
    );
    let mut events = fx.manager.events();
    fx.manager.start_liveness_watcher(Duration::from_millis(25));

    let idle = wait_for_activity(&fx.manager, session.id, ActivityState::Idle).await;
    assert!(
        idle.activity_since >= session.activity_since + chrono::Duration::seconds(2),
        "the Active -> Idle transition must stamp a fresh activity_since \
         (spawn stamp {:?}, transition stamp {:?})",
        session.activity_since,
        idle.activity_since
    );

    // Steady Idle across many more ticks must not re-stamp — no further
    // SessionUpdated at all, or the sidebar clock would keep resetting.
    while events.try_recv().is_ok() {}
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert!(
        matches!(events.try_recv(), Err(TryRecvError::Empty)),
        "a steady state must not re-emit (and re-stamp) on every tick"
    );
}

async fn wait_for_activity(
    manager: &SessionManager,
    id: uuid::Uuid,
    want: ActivityState,
) -> maestro_protocol::Session {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(6);
    loop {
        if tokio::time::Instant::now() >= deadline {
            panic!("timed out waiting for {id} to become {want:?}");
        }
        if let Some(s) = manager.snapshot().await.into_iter().find(|s| s.id == id) {
            if s.activity == want {
                return s;
            }
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

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

/// The kill handler used to check `has()` first, which opened a window for the
/// liveness watcher to reap the session in between — turning a 404 into a 502.
/// `kill` reports the miss itself now.
#[tokio::test]
async fn killing_an_unknown_session_is_reported_as_not_found() {
    let fx = Fx::new();
    let err = fx
        .manager
        .kill(uuid::Uuid::new_v4(), false)
        .await
        .expect_err("should fail");
    assert_eq!(err.to_error_info().code, "not_found");
}

/// A construction-order mistake must be loud: a silently-ignored history leaves
/// the recent-dirs picker permanently empty.
#[test]
#[should_panic(expected = "before the SessionManager is cloned")]
fn attaching_spawn_history_after_cloning_panics() {
    let tmp = TempDir::new().expect("tempdir");
    let fake = Arc::new(FakePanes::new());
    let manager = SessionManager::new(
        fake,
        SidecarStore::new(tmp.path().join("sidecars")),
        WorktreeManager::new(Some(tmp.path().join("worktrees"))),
        tmp.path().join("fifos"),
    );
    let _clone = manager.clone();
    let _ = manager.with_spawn_history(maestro_daemon::spawn_history::SpawnHistory::new(
        tmp.path().join("history.jsonl"),
        7,
    ));
}

/// Every tmux-spawned session learns who it is (#103): its own UUID under
/// `MAESTRO_SESSION_ID`, so the reporting hook in #106 can name the session
/// it is running inside.
#[tokio::test]
async fn spawn_exports_maestro_session_id() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");

    let env = fx.spawned_env();
    assert_eq!(
        env.get("MAESTRO_SESSION_ID").map(String::as_str),
        Some(&s.id.to_string()[..]),
        "MAESTRO_SESSION_ID must be the session's UUID, got {env:?}"
    );
}

/// The hook also needs to know where to report to: `MAESTRO_URL` carries the
/// daemon's base URL derived from `[daemon].bind_addr` (#103).
#[tokio::test]
async fn spawn_exports_maestro_url_when_the_daemon_is_wired() {
    let fx = Fx::with_daemon_url("http://127.0.0.1:7070");
    fx.manager.create(fx.request("bash")).await.expect("create");

    let env = fx.spawned_env();
    assert_eq!(
        env.get("MAESTRO_URL").map(String::as_str),
        Some("http://127.0.0.1:7070"),
        "MAESTRO_URL must be the daemon's base URL, got {env:?}"
    );
}

/// Without a wired URL (tests, or a caller that never set one) only the
/// session id is injected — an empty or bogus `MAESTRO_URL` would send the
/// #106 hook's reports nowhere useful.
#[tokio::test]
async fn spawn_omits_maestro_url_when_not_wired() {
    let fx = Fx::new();
    fx.manager.create(fx.request("bash")).await.expect("create");

    let env = fx.spawned_env();
    assert!(
        !env.contains_key("MAESTRO_URL"),
        "MAESTRO_URL must be absent when no daemon URL is known, got {env:?}"
    );
}

/// The daemon-owned variables win over anything the spawner sent: a client
/// cannot spoof another session's identity.
#[tokio::test]
async fn spawn_overrides_client_supplied_maestro_variables() {
    let fx = Fx::with_daemon_url("http://127.0.0.1:7070");
    let mut req = fx.request("bash");
    req.env.insert(
        "MAESTRO_SESSION_ID".into(),
        "00000000-0000-0000-0000-000000000000".into(),
    );
    req.env.insert("MAESTRO_URL".into(), "http://evil:1".into());
    let s = fx.manager.create(req).await.expect("create");

    let env = fx.spawned_env();
    assert_eq!(env.get("MAESTRO_SESSION_ID"), Some(&s.id.to_string()));
    assert_eq!(
        env.get("MAESTRO_URL"),
        Some(&"http://127.0.0.1:7070".to_string())
    );
}

/// Two clients attached to one session used to resize the shared window past
/// each other on every report they made. Every flip repaints a full-screen
/// agent at a new height and orphans the rows it drew at the old one, which
/// reads to the operator as a pane that has stopped taking input.
#[tokio::test]
async fn clients_of_different_sizes_do_not_resize_the_window_past_each_other() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");
    let web = fx.manager.attach_viewport();
    let tui = fx.manager.attach_viewport();

    fx.manager.set_viewport(s.id, web, 53, 46).await.unwrap();
    fx.manager.set_viewport(s.id, tui, 53, 20).await.unwrap();
    // Both keep reporting, as a browser's ResizeObserver and the TUI's own
    // redraw loop do. Neither report changes the arbitrated size.
    for _ in 0..5 {
        fx.manager.set_viewport(s.id, web, 53, 46).await.unwrap();
        fx.manager.set_viewport(s.id, tui, 53, 20).await.unwrap();
    }

    let resizes: Vec<_> = fx
        .fake
        .calls()
        .into_iter()
        .filter_map(|c| match c {
            Call::ResizeWindow { cols, rows, .. } => Some((cols, rows)),
            _ => None,
        })
        .collect();
    assert_eq!(
        resizes,
        vec![(53, 46), (53, 20)],
        "the window was resized more than the arbitrated size actually changed"
    );
}

/// The smaller client leaving hands the window back to the one still attached.
#[tokio::test]
async fn releasing_a_client_resizes_to_what_is_left() {
    let fx = Fx::new();
    let s = fx.manager.create(fx.request("bash")).await.expect("create");
    let web = fx.manager.attach_viewport();
    let tui = fx.manager.attach_viewport();
    fx.manager.set_viewport(s.id, web, 80, 46).await.unwrap();
    fx.manager.set_viewport(s.id, tui, 80, 20).await.unwrap();

    fx.manager.release_viewport(s.id, tui).await;

    let resizes: Vec<_> = fx
        .fake
        .calls()
        .into_iter()
        .filter_map(|c| match c {
            Call::ResizeWindow { cols, rows, .. } => Some((cols, rows)),
            _ => None,
        })
        .collect();
    assert_eq!(resizes, vec![(80, 46), (80, 20), (80, 46)]);
}