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

//! End-to-end tests for `POST /api/sessions/:id/resume` (#166) and the
//! manager call behind it, against real tmux.
//!
//! The point of resume is that the session comes back under the *same* uuid
//! and sidecar — name, `created_at`, env and worktree ownership intact — with
//! only the tmux session and the PTY bridge new.

mod common;

use std::collections::HashMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;

use common::{wait_for_async, DaemonFixture};
use maestro_daemon::{local_api, SessionError, SessionManager};
use maestro_protocol::{CreateSessionRequest, Session, SessionStatus, SpawnPreset};
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use uuid::Uuid;

/// A fresh manager over the same tmux server, sidecars and fifos — what the
/// daemon builds after a restart.
fn revive(fx: &DaemonFixture) -> SessionManager {
    SessionManager::new(
        Arc::new(fx.tmux.tmux.clone()),
        maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
        maestro_daemon::worktree::WorktreeManager::new(None),
        fx.fifo_dir(),
    )
}

/// Create a session, then simulate the host reboot that leaves it dormant:
/// the tmux server dies, the sidecar stays, a new manager recovers.
async fn dormant_session(
    fx: &DaemonFixture,
    cwd: &TempDir,
    command: &str,
    name: &str,
) -> (SessionManager, Session) {
    let mut metadata = HashMap::new();
    metadata.insert(
        maestro_protocol::NAME_KEY.to_string(),
        serde_json::Value::String(name.into()),
    );
    let session = fx
        .manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: command.into(),
            env: HashMap::from([("MAESTRO_TEST_ENV".to_string(), "kept".to_string())]),
            metadata,
            worktree: None,
        })
        .await
        .expect("create");
    fx.tmux.tmux.kill_server().expect("kill tmux server");
    let revived = revive(fx);
    revived.recover().await.expect("recover");
    (revived, session)
}

fn marker_preset(command: &str, marker_dir: &TempDir) -> Vec<SpawnPreset> {
    vec![SpawnPreset {
        id: "agent".into(),
        label: "Agent".into(),
        command: command.into(),
        resume_command: Some(format!(
            "sh -c 'touch {}/resumed-{{agent_session}}; sleep 30'",
            marker_dir.path().display()
        )),
    }]
}

/// The keystone: a dormant session comes back running under its own id, with
/// its name, `created_at`, env and sidecar untouched.
#[tokio::test]
async fn resume_brings_a_dormant_session_back_under_the_same_id() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, original) = dormant_session(&fx, &cwd, "sleep 30", "long-runner").await;
    assert_eq!(
        manager.status(original.id).await,
        Some(SessionStatus::Dormant)
    );

    let resumed = manager.resume(original.id, &[]).await.expect("resume");

    assert_eq!(resumed.id, original.id, "same uuid");
    assert_eq!(resumed.status, SessionStatus::Running);
    assert_eq!(resumed.created_at, original.created_at);
    assert_eq!(resumed.cwd, original.cwd);
    assert_eq!(resumed.command, original.command);
    assert_eq!(resumed.env, original.env);
    assert_eq!(
        resumed
            .metadata
            .get(maestro_protocol::NAME_KEY)
            .and_then(|v| v.as_str()),
        Some("long-runner")
    );

    // tmux really holds it again, and the sidecar is the same file, updated.
    let live = fx.tmux.tmux.list_sessions().expect("list");
    assert_eq!(live, vec![original.id]);
    let stored = maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir())
        .read(original.id)
        .expect("read sidecar")
        .expect("sidecar still there");
    assert_eq!(stored.status, SessionStatus::Running);
    assert_eq!(stored.created_at, original.created_at);

    // And it is attachable again, which a dormant session never was.
    assert!(manager
        .subscribe_pty_with_replay(original.id)
        .await
        .is_some());
}

/// The worktree the session owns stays the session's, so killing the resumed
/// session still cleans the tree up.
#[tokio::test]
async fn resume_keeps_worktree_ownership() {
    let fx = DaemonFixture::new();
    let repo = TempDir::new().unwrap();
    common::init_repo(repo.path());

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

    fx.tmux.tmux.kill_server().expect("kill tmux server");
    let manager = revive(&fx);
    manager.recover().await.expect("recover");
    let resumed = manager.resume(session.id, &[]).await.expect("resume");

    assert_eq!(resumed.worktree, Some(worktree.clone()));
    assert_eq!(
        resumed.cwd, worktree.path,
        "resumed inside its own worktree"
    );
}

/// The command comes from the preset's `resume_command`, with the session's
/// own agent conversation id (#161) substituted in.
#[tokio::test]
async fn resume_runs_the_presets_resume_command() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let markers = TempDir::new().unwrap();
    let presets = marker_preset("sleep 30", &markers);

    let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "agentic").await;
    // The agent reported its conversation id before the reboot, so it is on
    // the sidecar and the resume template can be filled in.
    fx.manager
        .report_activity(
            session.id,
            maestro_protocol::ActivityState::Idle,
            None,
            Some("conv-42".into()),
        )
        .await
        .ok();
    // Re-read the sidecar the report updated.
    let manager = {
        drop(manager);
        let m = revive(&fx);
        m.recover().await.expect("recover");
        m
    };

    manager.resume(session.id, &presets).await.expect("resume");

    let marker = markers.path().join("resumed-conv-42");
    wait_for_async(Duration::from_secs(5), || {
        let marker = marker.clone();
        async move { marker.exists() }
    })
    .await;
}

/// A session that is already running has nothing to bring back, and
/// respawning would strand the pane that is already there.
#[tokio::test]
async fn resume_refuses_a_running_session() {
    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");

    let err = fx
        .manager
        .resume(session.id, &[])
        .await
        .expect_err("already running");
    assert!(matches!(err, SessionError::AlreadyRunning(id) if id == session.id));
    assert_eq!(err.to_error_info().code, "already_running");
}

/// Nothing can be resumed in a directory that no longer exists.
#[tokio::test]
async fn resume_refuses_when_the_cwd_is_gone() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "homeless").await;

    std::fs::remove_dir_all(cwd.path()).unwrap();

    let err = manager
        .resume(session.id, &[])
        .await
        .expect_err("cwd is gone");
    assert_eq!(err.to_error_info().code, "cwd_missing");
    assert_eq!(
        manager.status(session.id).await,
        Some(SessionStatus::Dormant),
        "a refused resume leaves the session exactly as it was"
    );
}

#[tokio::test]
async fn resume_of_an_unknown_session_is_not_found() {
    let fx = DaemonFixture::new();
    let err = fx
        .manager
        .resume(Uuid::new_v4(), &[])
        .await
        .expect_err("unknown");
    assert_eq!(err.to_error_info().code, "not_found");
}

// ---------------------------------------------------------------------------
// HTTP surface
// ---------------------------------------------------------------------------

struct Harness {
    addr: SocketAddr,
    _shutdown: oneshot::Sender<()>,
}

async fn serve(manager: SessionManager, presets: Vec<SpawnPreset>) -> Harness {
    let router = local_api::router(local_api::state_for(
        manager,
        "test-host",
        "/home/test",
        presets,
        Arc::new(maestro_daemon::projects::ProjectScanner::disabled()),
        maestro_config::DEFAULT_LEADER,
    ));
    let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
        .await
        .unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, rx) = oneshot::channel::<()>();
    tokio::spawn(async move {
        let _ = axum::serve(listener, router)
            .with_graceful_shutdown(async move {
                let _ = rx.await;
            })
            .await;
    });
    Harness {
        addr,
        _shutdown: tx,
    }
}

#[tokio::test]
async fn post_resume_returns_the_revived_session() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "web-resumed").await;
    let h = serve(manager, Vec::new()).await;

    let resp = reqwest::Client::new()
        .post(format!(
            "http://{}{}",
            h.addr,
            maestro_protocol::routes::session_resume(session.id)
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::OK);
    let body: Session = resp.json().await.unwrap();
    assert_eq!(body.id, session.id);
    assert_eq!(body.status, SessionStatus::Running);
}

#[tokio::test]
async fn post_resume_on_a_running_session_is_a_conflict() {
    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");
    let h = serve(fx.manager.clone(), Vec::new()).await;

    let resp = reqwest::Client::new()
        .post(format!(
            "http://{}{}",
            h.addr,
            maestro_protocol::routes::session_resume(session.id)
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::CONFLICT);
    let err: maestro_protocol::ApiError = resp.json().await.unwrap();
    assert_eq!(err.error.code, "already_running");
}

#[tokio::test]
async fn post_resume_on_an_unknown_session_is_404() {
    let fx = DaemonFixture::new();
    let h = serve(fx.manager.clone(), Vec::new()).await;
    let resp = reqwest::Client::new()
        .post(format!(
            "http://{}{}",
            h.addr,
            maestro_protocol::routes::session_resume(Uuid::new_v4())
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
}

// ---------------------------------------------------------------------------
// Races (#175)
// ---------------------------------------------------------------------------

/// Real tmux, with the *first* `new_session_piped` held inside the call until
/// the test releases it. Later calls pass straight through, so a second resume
/// racing the first never blocks on a gate the first one owns.
///
/// This is what makes the two races below deterministic rather than timing
/// dependent: the resume under test is parked in exactly the window the bug
/// lives in, and the racing operation runs to completion before it moves.
#[derive(Debug)]
struct GatedSpawn {
    inner: maestro_daemon::tmux::Tmux,
    entered: tokio::sync::mpsc::UnboundedSender<()>,
    release: std::sync::Mutex<std::sync::mpsc::Receiver<()>>,
    gated: std::sync::atomic::AtomicBool,
}

/// The test's end of a [`GatedSpawn`].
struct Gate {
    entered: tokio::sync::mpsc::UnboundedReceiver<()>,
    release: std::sync::mpsc::SyncSender<()>,
}

impl Gate {
    /// Wait until the gated spawn is parked mid-resume.
    async fn wait_entered(&mut self) {
        tokio::time::timeout(Duration::from_secs(10), self.entered.recv())
            .await
            .expect("resume reached the tmux spawn")
            .expect("gate alive");
    }

    fn release(&self) {
        self.release.send(()).expect("gate receiver alive");
    }
}

fn gated_panes(inner: maestro_daemon::tmux::Tmux) -> (Arc<GatedSpawn>, Gate) {
    let (entered_tx, entered_rx) = tokio::sync::mpsc::unbounded_channel();
    let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1);
    (
        Arc::new(GatedSpawn {
            inner,
            entered: entered_tx,
            release: std::sync::Mutex::new(release_rx),
            gated: std::sync::atomic::AtomicBool::new(true),
        }),
        Gate {
            entered: entered_rx,
            release: release_tx,
        },
    )
}

impl maestro_daemon::panes::Panes for GatedSpawn {
    fn new_session_piped(
        &self,
        id: Uuid,
        cwd: &std::path::Path,
        command: &str,
        session_env: &std::collections::BTreeMap<String, String>,
        sink_command: &str,
    ) -> Result<(), maestro_daemon::tmux::TmuxError> {
        if self.gated.swap(false, std::sync::atomic::Ordering::SeqCst) {
            let _ = self.entered.send(());
            let _ = self.release.lock().expect("gate mutex").recv();
        }
        self.inner
            .new_session_piped(id, cwd, command, session_env, sink_command)
    }

    fn kill_session(&self, id: Uuid) -> Result<(), maestro_daemon::tmux::TmuxError> {
        self.inner.kill_session(id)
    }

    fn list_sessions(&self) -> Result<Vec<Uuid>, maestro_daemon::tmux::TmuxError> {
        self.inner.list_sessions()
    }

    fn pane_info(
        &self,
        id: Uuid,
    ) -> Result<Option<maestro_daemon::tmux::PaneInfo>, maestro_daemon::tmux::TmuxError> {
        self.inner.pane_info(id)
    }

    fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), maestro_daemon::tmux::TmuxError> {
        self.inner.send_keys(id, bytes)
    }

    fn pipe_pane(&self, id: Uuid, sink: &str) -> Result<(), maestro_daemon::tmux::TmuxError> {
        self.inner.pipe_pane(id, sink)
    }

    fn resize_window(
        &self,
        id: Uuid,
        cols: u16,
        rows: u16,
    ) -> Result<(), maestro_daemon::tmux::TmuxError> {
        self.inner.resize_window(id, cols, rows)
    }

    fn capture_pane(&self, id: Uuid) -> Result<Option<Vec<u8>>, maestro_daemon::tmux::TmuxError> {
        self.inner.capture_pane(id)
    }
}

/// A dormant session whose manager's tmux spawn can be parked by the test.
async fn dormant_gated(fx: &DaemonFixture, cwd: &TempDir) -> (SessionManager, Session, Gate) {
    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");
    fx.tmux.tmux.kill_server().expect("kill tmux server");

    let (panes, gate) = gated_panes(fx.tmux.tmux.clone());
    let manager = SessionManager::new(
        panes,
        maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
        maestro_daemon::worktree::WorktreeManager::new(None),
        fx.fifo_dir(),
    );
    manager.recover().await.expect("recover");
    assert_eq!(
        manager.status(session.id).await,
        Some(SessionStatus::Dormant)
    );
    (manager, session, gate)
}

/// A kill that lands while a resume is inside the tmux spawn must not leave
/// the sidecar behind. The kill deletes it; the resume writes it back a moment
/// later and then discovers its own record is gone. The sidecar it resurrected
/// would otherwise make the next recovery register a ghost dormant session for
/// a session that was killed.
#[tokio::test]
async fn a_kill_during_resume_leaves_no_sidecar_behind() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await;
    let id = session.id;

    let resuming = tokio::spawn({
        let manager = manager.clone();
        async move { manager.resume(id, &[]).await }
    });
    gate.wait_entered().await;

    manager.kill(id, false).await.expect("kill mid-resume");
    gate.release();

    let err = resuming.await.unwrap().expect_err("record is gone");
    assert!(matches!(err, SessionError::NotFound(got) if got == id));

    assert!(
        maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir())
            .read(id)
            .expect("read sidecar")
            .is_none(),
        "the killed session's sidecar must not be resurrected by the losing resume"
    );

    // And so the next recovery registers no ghost.
    let after = revive(&fx);
    after.recover().await.expect("recover");
    assert!(
        after.snapshot().await.is_empty(),
        "a killed session must not come back dormant"
    );
}

/// Two resumes of the same session must not both proceed. The second is
/// refused on the state — `already_running` — not on its tmux spawn colliding
/// with the first one's.
#[tokio::test]
async fn a_second_resume_is_refused_while_the_first_is_in_flight() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await;
    let id = session.id;

    let resuming = tokio::spawn({
        let manager = manager.clone();
        async move { manager.resume(id, &[]).await }
    });
    gate.wait_entered().await;

    let err = manager
        .resume(id, &[])
        .await
        .expect_err("a resume is already in flight");
    assert!(matches!(err, SessionError::AlreadyRunning(got) if got == id));
    assert_eq!(err.to_error_info().code, "already_running");

    gate.release();
    let first = resuming.await.unwrap().expect("the first resume wins");
    assert_eq!(first.status, SessionStatus::Running);
    assert_eq!(
        fx.tmux.tmux.list_sessions().expect("list"),
        vec![id],
        "exactly one pane for the session"
    );
}

/// A refused resume must not leave the session marked in-flight: once the
/// reason for the refusal is gone, resuming works.
#[tokio::test]
async fn a_refused_resume_can_be_retried() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "retried").await;

    std::fs::remove_dir_all(cwd.path()).unwrap();
    let err = manager
        .resume(session.id, &[])
        .await
        .expect_err("cwd is gone");
    assert_eq!(err.to_error_info().code, "cwd_missing");

    std::fs::create_dir_all(cwd.path()).unwrap();
    let resumed = manager.resume(session.id, &[]).await.expect("resume");
    assert_eq!(resumed.status, SessionStatus::Running);
}