diff --git a/crates/maestro-daemon/src/cli.rs b/crates/maestro-daemon/src/cli.rs index 35fb078..aeadad3 100644 --- a/crates/maestro-daemon/src/cli.rs +++ b/crates/maestro-daemon/src/cli.rs @@ -84,6 +84,9 @@ async fn run(config_override: Option) -> Result<(), Box> { config.spawn_history_path(), config.spawn_history_days, )) + // #171: mirror PTY scrollback so a resumed session opens on what it was + // showing, instead of a blank pane. + .with_scrollback(config.scrollback_dir()) .with_boot_id( config.boot_id_path(), std::sync::Arc::new(crate::boot_id::ProcBootId), diff --git a/crates/maestro-daemon/src/config.rs b/crates/maestro-daemon/src/config.rs index ffd849a..0376d3c 100644 --- a/crates/maestro-daemon/src/config.rs +++ b/crates/maestro-daemon/src/config.rs @@ -117,6 +117,12 @@ impl DaemonConfig { self.state_dir.join("spawn-history.jsonl") } + /// Where PTY scrollback is mirrored so a resumed session opens on what + /// it printed before (#171): `/scrollback`. + pub fn scrollback_dir(&self) -> PathBuf { + self.state_dir.join("scrollback") + } + /// Where recovery records the kernel boot id it saw last (#164): /// `/boot_id`. pub fn boot_id_path(&self) -> PathBuf { diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 2ea4aa1..0df7f5d 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -26,6 +26,7 @@ pub mod projects; pub mod pty; pub mod release; pub mod resume; +pub mod scrollback; pub mod session; pub mod sidecar; pub mod spawn_history; diff --git a/crates/maestro-daemon/src/pty.rs b/crates/maestro-daemon/src/pty.rs index 44f079d..e6d194f 100644 --- a/crates/maestro-daemon/src/pty.rs +++ b/crates/maestro-daemon/src/pty.rs @@ -21,6 +21,11 @@ //! one-shot prefix before joining the live broadcast, so attaching to an //! idle session immediately shows the last screenful instead of a blank //! terminal. +//! - #171: a bridge can be started with scrollback restored from disk — what +//! the session printed before the daemon (or the host) went away. Those +//! bytes are handed to subscribers ahead of the live ones until the ring +//! has a full budget of new output of its own, at which point the restored +//! history is older than anything the budget covers and is dropped. use std::collections::VecDeque; use std::ffi::CString; @@ -58,6 +63,10 @@ const READ_BUF_BYTES: usize = 4096; #[derive(Debug)] pub struct PtyBridge { fifo_path: PathBuf, + /// Scrollback from a previous life of this session (#171), replayed + /// ahead of the ring. Empty for a fresh spawn. Never appended to, so it + /// can never duplicate a byte the ring also holds. + restored: Vec, /// `None` once [`Self::seal`] has run: live subscribers see the channel /// close, but the replay buffer stays readable. sender: Option>>, @@ -79,6 +88,21 @@ impl PtyBridge { /// Variant that lets the caller pick a custom replay-buffer cap (bytes). /// Pass `0` to disable replay (subscribers see only live bytes). pub async fn start_with_replay(fifo_path: PathBuf, replay_bytes: usize) -> io::Result { + Self::start_restoring(fifo_path, replay_bytes, Vec::new()).await + } + + /// As [`Self::start_with_replay`], seeded with `restored` scrollback read + /// back from disk (#171). The seed is truncated to the same + /// `replay_bytes` budget the ring uses — there is one budget here, not + /// two — and is dropped entirely when replay is disabled. + pub async fn start_restoring( + fifo_path: PathBuf, + replay_bytes: usize, + mut restored: Vec, + ) -> io::Result { + if restored.len() > replay_bytes { + restored.drain(..restored.len() - replay_bytes); + } if let Some(parent) = fifo_path.parent() { tokio::fs::create_dir_all(parent).await?; } @@ -102,6 +126,7 @@ impl PtyBridge { }); Ok(Self { fifo_path, + restored, sender: Some(sender), replay, last_activity, @@ -155,7 +180,8 @@ impl PtyBridge { F: FnOnce() -> Option>, { let guard = self.replay.lock_for_snapshot(); - let snapshot = capture().unwrap_or_else(|| guard.snapshot()); + let live = capture().unwrap_or_else(|| guard.snapshot()); + let snapshot = guard.with_restored_prefix(&self.restored, live); let rx = match &self.sender { Some(tx) => tx.subscribe(), // Sealed: hand back the scrollback plus a receiver that is already @@ -171,6 +197,25 @@ impl PtyBridge { (snapshot, rx) } + /// Everything worth persisting for a later life of this session (#171): + /// the restored scrollback and the ring, trimmed together to the replay + /// budget so the file on disk is never bigger than the ring it mirrors. + pub fn scrollback(&self) -> Vec { + let guard = self.replay.lock_for_snapshot(); + guard.with_restored_prefix(&self.restored, guard.snapshot()) + } + + /// Total bytes this bridge has appended to its ring. The scrollback flush + /// pass compares it against what it last wrote, so an idle session costs + /// no disk writes at all. + pub fn bytes_seen(&self) -> u64 { + self.replay + .inner + .lock() + .expect("ReplayBuffer poisoned") + .seen + } + /// Stop reading and close the live stream, keeping the replay buffer. /// /// Attached clients see the channel close — without that a `maestro attach` @@ -201,6 +246,8 @@ struct ReplayBuffer { struct ReplayInner { buf: VecDeque, cap: usize, + /// Bytes ever appended, not bytes held — the ring evicts, this does not. + seen: u64, } impl ReplayBuffer { @@ -209,6 +256,7 @@ impl ReplayBuffer { inner: Arc::new(Mutex::new(ReplayInner { buf: VecDeque::with_capacity(cap.min(READ_BUF_BYTES * 2)), cap, + seen: 0, })), } } @@ -225,6 +273,7 @@ impl ReplayBuffer { // If the incoming chunk alone exceeds the cap, take only its tail — // anything in the buffer would be evicted before this finishes anyway. let take_from = bytes.len().saturating_sub(s.cap); + s.seen += bytes.len() as u64; let bytes = &bytes[take_from..]; s.buf.extend(bytes); while s.buf.len() > s.cap { @@ -247,6 +296,24 @@ struct ReplayGuard<'a> { } impl ReplayGuard<'_> { + /// `live` with the restored scrollback (#171) in front of it, while the + /// ring still holds less than a full budget of its own output. Once it + /// holds a full budget, everything restored is older than what the + /// budget covers, so it is dropped rather than kept forever. + fn with_restored_prefix(&self, restored: &[u8], live: Vec) -> Vec { + let cap = self.inner.cap; + let held = self.inner.buf.len(); + if restored.is_empty() || cap == 0 || held >= cap { + return live; + } + let keep = cap - held; + let start = restored.len().saturating_sub(keep); + let mut out = Vec::with_capacity((restored.len() - start) + live.len()); + out.extend_from_slice(&restored[start..]); + out.extend_from_slice(&live); + out + } + fn snapshot(&self) -> Vec { let (a, b) = self.inner.buf.as_slices(); let mut out = Vec::with_capacity(a.len() + b.len()); @@ -292,6 +359,7 @@ async fn run_reader( // never both, never neither. let mut s = replay.inner.lock().expect("ReplayBuffer poisoned"); if s.cap > 0 { + s.seen += n as u64; let take_from = n.saturating_sub(s.cap); s.buf.extend(&buf[take_from..n]); while s.buf.len() > s.cap { @@ -345,6 +413,80 @@ fn shell_quote(s: &str) -> String { mod tests { use super::*; + async fn bridge(cap: usize, restored: &[u8]) -> (tempfile::TempDir, PtyBridge) { + let dir = tempfile::TempDir::new().expect("tempdir"); + let bridge = + PtyBridge::start_restoring(dir.path().join("pty.fifo"), cap, restored.to_vec()) + .await + .expect("start bridge"); + (dir, bridge) + } + + /// Feed the bridge through its own FIFO and wait for the reader to have + /// taken the bytes, so the ring state is deterministic. + async fn feed(bridge: &PtyBridge, bytes: &[u8]) { + use std::io::Write; + let before = bridge.bytes_seen(); + let mut f = std::fs::OpenOptions::new() + .write(true) + .open(bridge.fifo_path()) + .expect("open fifo for write"); + f.write_all(bytes).expect("write fifo"); + drop(f); + for _ in 0..400 { + if bridge.bytes_seen() >= before + bytes.len() as u64 { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + panic!("bridge never read the bytes written to its fifo"); + } + + #[tokio::test] + async fn restored_scrollback_is_replayed_ahead_of_live_bytes() { + let (_dir, bridge) = bridge(64, b"before the reboot\n").await; + feed(&bridge, b"after\n").await; + let (replay, _rx) = bridge.subscribe_with_replay(); + assert_eq!(replay, b"before the reboot\nafter\n"); + } + + #[tokio::test] + async fn restored_scrollback_is_trimmed_to_the_replay_budget() { + let (_dir, bridge) = bridge(4, b"abcdefghij").await; + let (replay, _rx) = bridge.subscribe_with_replay(); + assert_eq!(replay, b"ghij"); + } + + #[tokio::test] + async fn restored_scrollback_yields_once_the_ring_holds_a_full_budget() { + let (_dir, bridge) = bridge(4, b"old!").await; + feed(&bridge, b"ab").await; + let (replay, _rx) = bridge.subscribe_with_replay(); + assert_eq!(replay, b"d!ab", "prefix plus ring stays inside the budget"); + feed(&bridge, b"cd").await; + let (replay, _rx) = bridge.subscribe_with_replay(); + assert_eq!( + replay, b"abcd", + "restored history is now older than the budget" + ); + } + + #[tokio::test] + async fn scrollback_carries_the_restored_prefix_and_the_ring() { + let (_dir, bridge) = bridge(8, b"old").await; + feed(&bridge, b"new").await; + assert_eq!(bridge.scrollback(), b"oldnew"); + assert_eq!(bridge.bytes_seen(), 3); + } + + #[tokio::test] + async fn no_replay_budget_means_no_restored_scrollback() { + let (_dir, bridge) = bridge(0, b"old").await; + let (replay, _rx) = bridge.subscribe_with_replay(); + assert!(replay.is_empty()); + assert!(bridge.scrollback().is_empty()); + } + #[test] fn replay_buffer_keeps_last_cap_bytes() { let r = ReplayBuffer::new(4); diff --git a/crates/maestro-daemon/src/scrollback.rs b/crates/maestro-daemon/src/scrollback.rs new file mode 100644 index 0000000..a8a1dd5 --- /dev/null +++ b/crates/maestro-daemon/src/scrollback.rs @@ -0,0 +1,148 @@ +//! Persisted PTY scrollback (#171). +//! +//! The PTY replay ring lives in memory, so everything a session printed is +//! gone the moment the daemon process is. A resumed session then opens on a +//! blank pane: the agent's conversation is back, but the terminal history the +//! operator was reading is not. +//! +//! This store mirrors each bridge's ring to `/scrollback/.bin`, +//! and hands it back when a session gets a new bridge — on resume, and on the +//! recovery pass that re-adopts a session tmux still holds. The file is +//! capped at the same `replay_bytes` budget the in-memory ring uses, so disk +//! cost is bounded by the same number the operator already configures. +//! +//! Nothing here is allowed to fail a resume. The payload is raw PTY bytes +//! with no framing to be corrupt, and a file that is missing, unreadable or +//! oversized degrades to an empty (or truncated) pane rather than an error. + +use std::io; +use std::path::{Path, PathBuf}; + +use uuid::Uuid; + +/// Filesystem-backed scrollback store, rooted at a directory. +#[derive(Debug, Clone)] +pub struct ScrollbackStore { + dir: PathBuf, +} + +impl ScrollbackStore { + pub fn new(dir: impl Into) -> Self { + Self { dir: dir.into() } + } + + pub fn dir(&self) -> &Path { + &self.dir + } + + pub fn path_for(&self, id: Uuid) -> PathBuf { + self.dir.join(format!("{id}.bin")) + } + + /// Atomically write `bytes` to `/.bin`, creating the directory if + /// needed. Written through a temp file and renamed, so a reboot in the + /// middle of a flush leaves the previous scrollback intact rather than a + /// half-file. + pub fn write(&self, id: Uuid, bytes: &[u8]) -> io::Result<()> { + std::fs::create_dir_all(&self.dir)?; + let tmp = self.dir.join(format!("{id}.bin.tmp")); + std::fs::write(&tmp, bytes)?; + std::fs::rename(&tmp, self.path_for(id)) + } + + /// The stored scrollback for `id`, capped to the last `budget` bytes. + /// + /// Never fails: a missing file, an unreadable one, or a `budget` of zero + /// all read as "no scrollback", which is exactly the blank pane the + /// operator had before this existed. + pub fn read(&self, id: Uuid, budget: usize) -> Vec { + if budget == 0 { + return Vec::new(); + } + let path = self.path_for(id); + let mut bytes = match std::fs::read(&path) { + Ok(b) => b, + Err(e) if e.kind() == io::ErrorKind::NotFound => return Vec::new(), + Err(e) => { + tracing::warn!(error = ?e, %id, path = %path.display(), "scrollback: unreadable, resuming with a blank pane"); + return Vec::new(); + } + }; + // A file written under a larger budget than this daemon now runs + // with keeps its tail, the same end of the stream the ring would. + if bytes.len() > budget { + bytes.drain(..bytes.len() - budget); + } + bytes + } + + /// Delete the scrollback for `id`. No-op if it is not there. + pub fn delete(&self, id: Uuid) -> io::Result<()> { + match std::fs::remove_file(self.path_for(id)) { + Ok(()) => Ok(()), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(e), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn store() -> (tempfile::TempDir, ScrollbackStore) { + let dir = tempfile::TempDir::new().expect("tempdir"); + let store = ScrollbackStore::new(dir.path().join("scrollback")); + (dir, store) + } + + #[test] + fn round_trips_bytes() { + let (_dir, store) = store(); + let id = Uuid::new_v4(); + store.write(id, b"hello world").expect("write"); + assert_eq!(store.read(id, 64), b"hello world"); + } + + #[test] + fn a_missing_file_reads_as_empty() { + let (_dir, store) = store(); + assert!(store.read(Uuid::new_v4(), 64).is_empty()); + } + + #[test] + fn a_file_over_budget_keeps_its_tail() { + let (_dir, store) = store(); + let id = Uuid::new_v4(); + store.write(id, b"abcdefghij").expect("write"); + assert_eq!(store.read(id, 4), b"ghij"); + } + + #[test] + fn a_zero_budget_reads_as_empty() { + let (_dir, store) = store(); + let id = Uuid::new_v4(); + store.write(id, b"abcdefghij").expect("write"); + assert!(store.read(id, 0).is_empty()); + } + + #[test] + fn delete_removes_the_file_and_tolerates_a_missing_one() { + let (_dir, store) = store(); + let id = Uuid::new_v4(); + store.write(id, b"gone soon").expect("write"); + store.delete(id).expect("delete"); + assert!(!store.path_for(id).exists()); + store.delete(id).expect("second delete is a no-op"); + } + + #[test] + fn a_second_write_replaces_the_first() { + let (_dir, store) = store(); + let id = Uuid::new_v4(); + store.write(id, b"first").expect("write"); + store.write(id, b"second").expect("rewrite"); + assert_eq!(store.read(id, 64), b"second"); + assert!(!store.dir().join(format!("{id}.bin.tmp")).exists()); + } +} diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index ccd4afd..321bf53 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -23,6 +23,7 @@ use uuid::Uuid; use crate::boot_id::BootIdSource; use crate::panes::Panes; use crate::pty::{pipe_pane_command, PtyBridge}; +use crate::scrollback::ScrollbackStore; use crate::sidecar::{SidecarError, SidecarStore}; use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord}; use crate::tmux::TmuxError; @@ -44,6 +45,16 @@ pub const DEFAULT_LIVENESS_INTERVAL: Duration = Duration::from_secs(2); /// attempt, which is all #83 needs. pub const DEFAULT_BLOCKED_WORK_INTERVAL: Duration = Duration::from_secs(15); +/// How often the liveness watcher mirrors PTY scrollback to disk (#171). +/// +/// A host reboot gives no warning and runs no exit path, so this interval is +/// the whole guarantee: whatever a session printed in the last few seconds +/// before the power went is what a resumed pane is missing. Ten seconds keeps +/// that loss cosmetic — scrollback is visual continuity, not state — while +/// costing at most one file write per *active* session per tick, since a +/// session whose bridge read nothing is skipped entirely. +pub const DEFAULT_SCROLLBACK_INTERVAL: Duration = Duration::from_secs(10); + /// Gap between the text of a `send` and the Enter that submits it. /// /// Agents that run a full-screen prompt (Claude Code, opencode) coalesce a @@ -97,6 +108,10 @@ pub struct SessionRecord { /// glow, this holds through focus loss until the session does new work. /// Daemon-local; never rides the wire. pub awaiting_suppressed: bool, + /// [`PtyBridge::bytes_seen`] as of this session's last scrollback flush + /// (#171). The flush pass skips a session whose bridge has read nothing + /// since, so an idle session costs no disk writes. + pub flushed_bytes: u64, } #[derive(Debug, thiserror::Error)] @@ -206,6 +221,10 @@ struct Inner { /// [`ResumeOnBoot::Dormant`] with no presets: register and wait. resume_policy: ResumeOnBoot, resume_presets: Vec, + /// Where PTY scrollback is mirrored so it survives the daemon, and the + /// host (#171). `None` — the default — means scrollback is not persisted + /// at all: a resumed session opens on a blank pane, exactly as before. + scrollback: Option, } /// Where recovery persists the boot id it last saw, and how it reads the @@ -266,6 +285,7 @@ impl SessionManager { boot_id: None, resume_policy: ResumeOnBoot::default(), resume_presets: Vec::new(), + scrollback: None, }), } } @@ -303,6 +323,21 @@ impl SessionManager { /// # Panics /// /// If the handle has already been cloned. + /// Persist PTY scrollback under `dir` (#171), so a resumed session comes + /// back showing what it printed before the daemon — or the host — went + /// away. Builder, called once right after construction like the other + /// `with_*` methods. Without it nothing is written, read or deleted. + /// + /// # Panics + /// + /// If the handle has already been cloned; see [`Self::with_spawn_history`]. + pub fn with_scrollback(mut self, dir: PathBuf) -> Self { + let inner = Arc::get_mut(&mut self.inner) + .expect("with_scrollback must be called before the SessionManager is cloned"); + inner.scrollback = Some(ScrollbackStore::new(dir)); + self + } + pub fn with_boot_id(mut self, path: PathBuf, source: Arc) -> Self { let inner = Arc::get_mut(&mut self.inner) .expect("with_boot_id must be called before the SessionManager is cloned"); @@ -601,6 +636,7 @@ impl SessionManager { focused_since: None, awaiting_suppressed: false, resuming: false, + flushed_bytes: 0, }, ); } @@ -676,8 +712,14 @@ impl SessionManager { // Same fifo path as the original spawn: it is derived from the id, and // `mkfifo` tolerates the stale node the previous run left behind. let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); - let bridge = - PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?; + // #171: the pane is new, but the scrollback the operator was reading + // before the reboot is on disk. It goes in ahead of the live bytes. + let bridge = PtyBridge::start_restoring( + fifo_path.clone(), + self.inner.replay_bytes, + self.restored_scrollback(id), + ) + .await?; let mut session_env = sanitize_env(&session.env)?; session_env.insert("MAESTRO_SESSION_ID".into(), id.to_string()); if let Some(url) = self @@ -756,6 +798,43 @@ impl SessionManager { Ok(session) } + /// Delete the scrollback of a sidecar recovery threw away (#171): the + /// session will never be listed again, so nothing would ever read it. + fn prune_scrollback(&self, id: Uuid) { + if let Some(store) = &self.inner.scrollback { + if let Err(e) = store.delete(id) { + tracing::warn!(error = ?e, %id, "recover: failed to prune scrollback"); + } + } + } + + /// Scrollback persisted for `id` by an earlier life of this session + /// (#171), capped to the replay budget. Empty when scrollback is not + /// wired, and empty — never an error — when the file is missing or + /// unreadable: a blank pane is the honest fallback, and it is what the + /// operator had before any of this existed. + fn restored_scrollback(&self, id: Uuid) -> Vec { + self.inner + .scrollback + .as_ref() + .map(|store| store.read(id, self.inner.replay_bytes)) + .unwrap_or_default() + } + + /// Mirror every live bridge's scrollback to disk (#171). + /// + /// Called on a timer by the liveness watcher rather than only on a clean + /// exit, because the case this exists for — a host reboot — runs no exit + /// path at all: anything flushed only on shutdown is missing exactly when + /// it is needed. Sessions whose bridge has read nothing since the last + /// flush are skipped, so an idle host writes nothing. + /// + /// Best-effort throughout: a failed write is warned and the session keeps + /// running. + pub async fn flush_scrollback(&self) { + run_scrollback_pass(&self.inner).await; + } + /// Drop the in-flight marker [`Self::resume`] set under its first lock /// (#175), so a refused or failed resume can be retried. A no-op when the /// record is already gone: a kill that landed mid-resume took the marker @@ -856,6 +935,10 @@ impl SessionManager { // Start "due" so the first live tick probes: a recovered daemon // should not wait a full interval to show what blocks. let mut since_probe = blocked_work_interval; + // Scrollback starts *not* due: a daemon that just recovered has + // nothing new to mirror, and the sessions it adopted were primed + // from the files this would overwrite (#171). + let mut since_flush = Duration::ZERO; loop { ticker.tick().await; let Some(inner) = weak.upgrade() else { @@ -870,6 +953,11 @@ impl SessionManager { since_probe = Duration::ZERO; run_blocked_work_pass(&inner).await; } + since_flush += interval; + if since_flush >= DEFAULT_SCROLLBACK_INTERVAL { + since_flush = Duration::ZERO; + run_scrollback_pass(&inner).await; + } } }); } @@ -929,8 +1017,15 @@ impl SessionManager { }, }; let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); - let bridge = - PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?; + // The ring the previous daemon held died with it, so an adopted + // session would otherwise replay nothing until tmux's own screen + // capture — which is one screenful, not scrollback (#171). + let bridge = PtyBridge::start_restoring( + fifo_path.clone(), + self.inner.replay_bytes, + self.restored_scrollback(id), + ) + .await?; // Re-pipe so output starts flowing again. `Tmux::pipe_pane` runs // without tmux's `-o` flag, so any stale pipe the previous daemon // left running to a now-unread FIFO is replaced (NAP-228). @@ -958,6 +1053,7 @@ impl SessionManager { focused_since: None, awaiting_suppressed: false, resuming: false, + flushed_bytes: 0, }, ); let _ = self @@ -1049,6 +1145,7 @@ impl SessionManager { if let Err(e) = self.inner.sidecar.delete(session.id) { tracing::warn!(error = ?e, id = %session.id, "recover: failed to prune sidecar with missing cwd"); } + self.prune_scrollback(session.id); pruned += 1; continue; } @@ -1059,6 +1156,7 @@ impl SessionManager { if let Err(e) = self.inner.sidecar.delete(session.id) { tracing::warn!(error = ?e, id = %session.id, "recover: failed to prune sidecar under resume_on_boot=never"); } + self.prune_scrollback(session.id); discarded += 1; continue; } @@ -1085,6 +1183,7 @@ impl SessionManager { focused_since: None, awaiting_suppressed: false, resuming: false, + flushed_bytes: 0, }, ); registered.push(session); @@ -1724,6 +1823,36 @@ async fn run_blocked_work_pass(inner: &Arc) { } } +/// Mirror every live bridge's scrollback to disk. See +/// [`SessionManager::flush_scrollback`]. +async fn run_scrollback_pass(inner: &Arc) { + let Some(store) = inner.scrollback.clone() else { + return; + }; + // Snapshot under the lock, write with it released: the writes are + // filesystem work and every attach needs this map. + let pending: Vec<(Uuid, Vec)> = { + let mut sessions = inner.sessions.lock().await; + sessions + .iter_mut() + .filter_map(|(id, rec)| { + let bridge = rec.bridge.as_ref()?; + let seen = bridge.bytes_seen(); + if seen == rec.flushed_bytes { + return None; + } + rec.flushed_bytes = seen; + Some((*id, bridge.scrollback())) + }) + .collect() + }; + for (id, bytes) in pending { + if let Err(e) = store.write(id, &bytes) { + tracing::warn!(error = ?e, %id, "scrollback: flush failed"); + } + } +} + /// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`] /// and settle its worktree per the natural-exit rule (#82). /// @@ -1750,6 +1879,15 @@ async fn mark_session_dead(inner: &Arc, id: Uuid) { // attach. if let Some(bridge) = record.bridge.as_mut() { bridge.seal(); + // The clean-exit half of #171: the last thing the session + // printed is what a later resume should open on, and it + // will produce nothing more for the flush pass to notice. + if let Some(store) = &inner.scrollback { + if let Err(e) = store.write(id, &bridge.scrollback()) { + tracing::warn!(error = ?e, %id, "scrollback: final flush failed"); + } + record.flushed_bytes = bridge.bytes_seen(); + } } let session = record.session.clone(); let worktree = record @@ -1871,6 +2009,14 @@ async fn cleanup_record( if let Err(e) = inner.sidecar.delete(id) { tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar"); } + // #171: the session is gone, so its scrollback goes with it — nothing + // will ever read the file again, and leaving it would keep a killed + // session's output on disk indefinitely. + if let Some(store) = &inner.scrollback { + if let Err(e) = store.delete(id) { + tracing::warn!(error = ?e, %id, "cleanup: failed to delete scrollback"); + } + } if let (Some(base), Some(path)) = (&record.worktree_base_repo, &record.worktree_path) { if keep_worktree { tracing::info!( diff --git a/crates/maestro-daemon/tests/common/mod.rs b/crates/maestro-daemon/tests/common/mod.rs index fc382d7..a6e0142 100644 --- a/crates/maestro-daemon/tests/common/mod.rs +++ b/crates/maestro-daemon/tests/common/mod.rs @@ -151,27 +151,40 @@ pub struct DaemonFixture { impl DaemonFixture { pub fn new() -> Self { - Self::build(None) + Self::build(None, maestro_daemon::pty::DEFAULT_REPLAY_BYTES) + } + + /// Variant whose manager runs a custom PTY replay budget (NAP-229), which + /// also caps the scrollback it persists (#171). + pub fn new_with_replay(replay_bytes: usize) -> Self { + Self::build(None, replay_bytes) } /// Variant whose manager has a durable spawn history (#24) attached, with /// `retention_days` recency cap, logging to `/spawn-history.jsonl`. pub fn new_with_history(retention_days: u32) -> Self { - Self::build(Some(retention_days)) + Self::build( + Some(retention_days), + maestro_daemon::pty::DEFAULT_REPLAY_BYTES, + ) } - fn build(history_days: Option) -> Self { + fn build(history_days: Option, replay_bytes: usize) -> Self { let tmux = TmuxFixture::new(); let state_dir = TempDir::new().unwrap(); let sidecar = SidecarStore::new(state_dir.path().join("sessions")); // Default placement: worktrees land under their base repo (#56). let worktree_mgr = WorktreeManager::new(None); - let mut manager = SessionManager::new( + let mut manager = SessionManager::new_with_replay( Arc::new(tmux.tmux.clone()), sidecar, worktree_mgr, state_dir.path().join("fifos"), + replay_bytes, ); + // #171: production wires this, so the fixtures do too — every test + // then exercises the same flush/restore/delete paths the daemon runs. + manager = manager.with_scrollback(state_dir.path().join("scrollback")); if let Some(days) = history_days { manager = manager.with_spawn_history(SpawnHistory::new(Self::history_path(&state_dir), days)); @@ -198,4 +211,8 @@ impl DaemonFixture { pub fn fifo_dir(&self) -> PathBuf { self.state_dir.path().join("fifos") } + + pub fn scrollback_dir(&self) -> PathBuf { + self.state_dir.path().join("scrollback") + } } diff --git a/crates/maestro-daemon/tests/scrollback.rs b/crates/maestro-daemon/tests/scrollback.rs new file mode 100644 index 0000000..e034737 --- /dev/null +++ b/crates/maestro-daemon/tests/scrollback.rs @@ -0,0 +1,219 @@ +//! 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" + ); +} diff --git a/docs/RESTART.md b/docs/RESTART.md index d85c1cc..e068e40 100644 --- a/docs/RESTART.md +++ b/docs/RESTART.md @@ -108,6 +108,32 @@ dormant ghost on the next recovery. The other thing to do with a dormant session is dismiss it, which removes the sidecar so the next recovery does not bring it back. +## What the pane looks like when it comes back + +The PTY replay ring lives in memory, so a resumed session would otherwise open +on a blank pane: the conversation is back, the terminal history the operator +was reading is not. Each bridge's ring is mirrored to +`/scrollback/.bin` (#171) and handed to the new bridge on +resume, and on the recovery pass that re-adopts a session tmux still holds — +that one lost its ring with the previous daemon too. The restored bytes go out +ahead of the live ones, and are dropped once the ring holds a full budget of +new output of its own, which is the point where everything restored is older +than the budget covers. + +The file is capped at the same `[daemon] pty_replay_bytes` budget the ring +uses — one budget, not two — and it is deleted when the session is killed, and +when recovery prunes its sidecar. Nothing here can fail a resume: a missing, +unreadable or oversized file degrades to an empty pane, which is exactly what +the operator had before it existed. + +It is flushed every 10 seconds by the liveness watcher, and once more when a +session's bridge is sealed on exit. The timer is the part that matters: a host +reboot runs no exit path at all, so scrollback written only on a clean shutdown +would be missing in the one case this exists for. Ten seconds bounds the loss +to the last few lines — scrollback is visual continuity, not state — and costs +at most one file write per *active* session per tick, since a session whose +bridge has read nothing since the last flush is skipped. + ## What recovery does with them: `resume_on_boot` Whether a recovered-but-dead session is left dormant at all is the operator's