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

//! Persisted PTY scrollback across a reboot (#171), against real tmux.
//!
//! Everything else in the reboot milestone restores what a session *is*; this
//! restores what it *looked like*. The cases that matter are the ones a clean
//! shutdown never gets to run: the host went down, and the only scrollback
//! that exists is whatever the last periodic flush wrote.

mod common;

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use common::{wait_for_async, DaemonFixture};
use maestro_daemon::scrollback::ScrollbackStore;
use maestro_daemon::SessionManager;
use maestro_protocol::{CreateSessionRequest, Session, SessionStatus};
use tempfile::TempDir;
use uuid::Uuid;

/// A fresh manager over the same tmux server, sidecars, fifos and scrollback —
/// 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(),
    )
    .with_scrollback(fx.scrollback_dir())
}

async fn spawn(fx: &DaemonFixture, cwd: &TempDir, command: &str) -> Session {
    fx.manager
        .create(CreateSessionRequest {
            cwd: cwd.path().to_path_buf(),
            command: command.into(),
            env: HashMap::new(),
            metadata: HashMap::new(),
            worktree: None,
        })
        .await
        .expect("create")
}

/// Wait until the session has printed `needle`, flushing as we go — the same
/// thing the liveness watcher's flush pass does on its timer.
async fn wait_for_flushed(
    manager: &SessionManager,
    store: &ScrollbackStore,
    id: Uuid,
    needle: &str,
) {
    wait_for_async(Duration::from_secs(10), || async {
        manager.flush_scrollback().await;
        String::from_utf8_lossy(&store.read(id, 64 * 1024)).contains(needle)
    })
    .await;
}

fn replay_text(replay: &[u8]) -> String {
    String::from_utf8_lossy(replay).to_string()
}

/// The keystone: the host reboots with no exit path run at all, and the
/// resumed session opens showing what it printed before.
#[tokio::test]
async fn a_resumed_session_replays_the_scrollback_from_before_the_reboot() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let session = spawn(&fx, &cwd, "sh -c 'echo before-the-reboot; sleep 30'").await;
    wait_for_flushed(&fx.manager, &store, session.id, "before-the-reboot").await;

    // The reboot: tmux dies where it stands. No kill, no seal, no shutdown
    // hook — only whatever the periodic flush had already written.
    fx.tmux.tmux.kill_server().expect("kill tmux server");
    let manager = revive(&fx);
    manager.recover().await.expect("recover");
    assert_eq!(
        manager.status(session.id).await,
        Some(SessionStatus::Dormant)
    );

    manager.resume(session.id, &[]).await.expect("resume");
    let (replay, _rx) = manager
        .subscribe_pty_with_replay(session.id)
        .await
        .expect("attachable again");
    assert!(
        replay_text(&replay).contains("before-the-reboot"),
        "resumed pane should open on the old scrollback, got {:?}",
        replay_text(&replay)
    );
}

/// A daemon restart leaves tmux alone, so the session is adopted rather than
/// resumed — and it too lost its in-memory ring with the old process.
#[tokio::test]
async fn an_adopted_session_replays_the_scrollback_of_the_previous_daemon() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let session = spawn(&fx, &cwd, "sh -c 'echo printed-earlier; sleep 30'").await;
    wait_for_flushed(&fx.manager, &store, session.id, "printed-earlier").await;

    // Daemon restart: the manager goes, tmux stays.
    let manager = revive(&fx);
    manager.recover().await.expect("recover");
    assert_eq!(
        manager.status(session.id).await,
        Some(SessionStatus::Running)
    );

    let (replay, _rx) = manager
        .subscribe_pty_with_replay(session.id)
        .await
        .expect("attachable");
    assert!(
        replay_text(&replay).contains("printed-earlier"),
        "adopted pane should carry the old scrollback ahead of tmux's screen, got {:?}",
        replay_text(&replay)
    );
}

/// Killing a session takes its scrollback with it: nothing will ever read the
/// file again, and the operator's output should not linger on disk.
#[tokio::test]
async fn killing_a_session_deletes_its_scrollback() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let session = spawn(&fx, &cwd, "sh -c 'echo doomed; sleep 30'").await;
    wait_for_flushed(&fx.manager, &store, session.id, "doomed").await;
    assert!(store.path_for(session.id).exists());

    fx.manager.kill(session.id, false).await.expect("kill");
    assert!(
        !store.path_for(session.id).exists(),
        "scrollback should not outlive the session"
    );
}

/// A session that exits on its own does get an exit path, and takes it: the
/// last thing it printed is flushed when the bridge is sealed, with no wait
/// for the next tick.
#[tokio::test]
async fn a_session_that_exits_flushes_its_last_output() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let session = spawn(&fx, &cwd, "sh -c 'echo last-words'").await;
    fx.manager
        .start_liveness_watcher(Duration::from_millis(100));
    wait_for_async(Duration::from_secs(10), || async {
        fx.manager.status(session.id).await == Some(SessionStatus::Dead)
    })
    .await;

    assert!(
        String::from_utf8_lossy(&store.read(session.id, 64 * 1024)).contains("last-words"),
        "a sealed bridge should have flushed what it printed"
    );
}

/// Nothing here may fail a resume. A scrollback file that is gone, or one full
/// of bytes no terminal would emit, degrades to a pane with no history — not
/// to an error.
#[tokio::test]
async fn a_missing_or_corrupt_scrollback_still_resumes() {
    let fx = DaemonFixture::new();
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let gone = spawn(&fx, &cwd, "sleep 30").await;
    let garbled = spawn(&fx, &cwd, "sleep 30").await;
    fx.manager.flush_scrollback().await;

    fx.tmux.tmux.kill_server().expect("kill tmux server");
    std::fs::create_dir_all(store.dir()).expect("scrollback dir");
    std::fs::remove_file(store.path_for(gone.id)).ok();
    std::fs::write(store.path_for(garbled.id), [0xff, 0xfe, 0x00, 0x80]).expect("write garbage");

    let manager = revive(&fx);
    manager.recover().await.expect("recover");
    for id in [gone.id, garbled.id] {
        manager.resume(id, &[]).await.expect("resume");
        assert_eq!(manager.status(id).await, Some(SessionStatus::Running));
        assert!(manager.subscribe_pty_with_replay(id).await.is_some());
    }
}

/// The disk cost is the replay budget, not a second one of its own.
#[tokio::test]
async fn the_flushed_file_stays_inside_the_replay_budget() {
    let fx = DaemonFixture::new_with_replay(256);
    let cwd = TempDir::new().unwrap();
    let store = ScrollbackStore::new(fx.scrollback_dir());

    let session = spawn(
        &fx,
        &cwd,
        "sh -c 'for i in $(seq 1 400); do echo line-$i; done; sleep 30'",
    )
    .await;
    wait_for_flushed(&fx.manager, &store, session.id, "line-400").await;

    let size = std::fs::metadata(store.path_for(session.id))
        .expect("stat scrollback")
        .len();
    assert!(
        size <= 256,
        "scrollback file was {size} bytes, budget is 256"
    );
}