diff --git a/crates/maestro-config/src/lib.rs b/crates/maestro-config/src/lib.rs index d8c429d..3624ca3 100644 --- a/crates/maestro-config/src/lib.rs +++ b/crates/maestro-config/src/lib.rs @@ -141,6 +141,11 @@ pub struct DaemonSection { /// back to peer daemons as a T2 dial target. #[serde(default, skip_serializing_if = "Option::is_none")] pub private_addr: Option, + /// Per-session PTY replay buffer cap in bytes (NAP-229). Each new attach + /// receives this much of the prior PTY output before the live stream. + /// `None` ⇒ daemon default (~64 KB); `Some(0)` disables replay. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pty_replay_bytes: Option, /// `[daemon.policy]` — per-daemon attach policy (NAP-247). #[serde(default, skip_serializing_if = "DaemonPolicySection::is_empty")] pub policy: DaemonPolicySection, @@ -156,6 +161,7 @@ impl DaemonSection { && self.worktree_root.is_none() && self.perimeter_id.is_none() && self.private_addr.is_none() + && self.pty_replay_bytes.is_none() && self.policy.is_empty() } } @@ -331,6 +337,7 @@ mod tests { worktree_root: Some(PathBuf::from("/var/lib/maestrod/wt")), perimeter_id: Some("wg-naps62".into()), private_addr: Some("10.0.0.7:7070".into()), + pty_replay_bytes: Some(131_072), policy: DaemonPolicySection { allow_inbound: vec!["konishi".into()], }, diff --git a/crates/maestro-daemon/src/attach.rs b/crates/maestro-daemon/src/attach.rs index 5b16a5c..c65c2e9 100644 --- a/crates/maestro-daemon/src/attach.rs +++ b/crates/maestro-daemon/src/attach.rs @@ -109,8 +109,8 @@ pub(crate) async fn serve_local( ) -> Result<(), axum::Error> { let (mut sink, mut stream) = socket.split(); - let mut rx = match manager.subscribe_pty(id).await { - Some(rx) => rx, + let (replay, mut rx) = match manager.subscribe_pty_with_replay(id).await { + Some(pair) => pair, None => { // session was killed between the auth check and now — let the // client see a clean close instead of a dropped socket. @@ -119,6 +119,14 @@ pub(crate) async fn serve_local( } }; + // NAP-229: send buffered scrollback up-front so the client sees the last + // screenful immediately instead of a blank terminal. The replay lock the + // bridge held during snapshot+subscribe guarantees no byte appears both + // here and on the live receiver, and none is dropped at the seam. + if !replay.is_empty() && sink.send(Message::Binary(replay)).await.is_err() { + return Ok(()); + } + // Set to true when the session itself terminates (broadcast closes); we // owe the client a Close frame in that case. Local breakouts (e.g. the // client sent its own Close, or the socket died) skip it. diff --git a/crates/maestro-daemon/src/config.rs b/crates/maestro-daemon/src/config.rs index 0e60d18..4e52aac 100644 --- a/crates/maestro-daemon/src/config.rs +++ b/crates/maestro-daemon/src/config.rs @@ -86,6 +86,10 @@ pub struct DaemonConfig { /// daemon-ids permitted to broker-attach to this daemon's sessions. /// Default empty (deny all). Local attach bypasses. (NAP-247.) pub allow_inbound: Vec, + /// Per-session PTY replay buffer cap in bytes (NAP-229). Each new attach + /// receives this much of the prior PTY output before the live stream. + /// Default `pty::DEFAULT_REPLAY_BYTES`; `0` disables. + pub pty_replay_bytes: usize, } impl DaemonConfig { @@ -198,6 +202,9 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result Result) -> Result<(), Box> { let sidecar = SidecarStore::new(config.sidecar_dir()); let worktree_mgr = WorktreeManager::new(config.worktree_root.clone()); let tmux = Tmux::new(); - let manager = SessionManager::new( + let manager = SessionManager::new_with_replay( config.daemon_id.clone(), tmux, sidecar, worktree_mgr, config.fifo_dir(), + config.pty_replay_bytes, ); let recovered = manager.recover().await?; diff --git a/crates/maestro-daemon/src/pty.rs b/crates/maestro-daemon/src/pty.rs index cb5a2ba..49c862d 100644 --- a/crates/maestro-daemon/src/pty.rs +++ b/crates/maestro-daemon/src/pty.rs @@ -16,13 +16,20 @@ //! down cleanly between tests. //! - We wrap the non-blocking file descriptor in [`AsyncFd`] and use //! `try_io` to read directly off the FD, leaving the blocking pool alone. +//! - NAP-229: every byte we broadcast is also appended to a per-bridge ring +//! buffer (default 64 KB). New subscribers receive the buffer as a +//! one-shot prefix before joining the live broadcast, so attaching to an +//! idle session immediately shows the last screenful instead of a blank +//! terminal. +use std::collections::VecDeque; use std::ffi::CString; use std::fs::File; use std::io::{self, Read}; use std::os::unix::ffi::OsStrExt; use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; use tokio::io::unix::AsyncFd; use tokio::io::Interest; @@ -34,6 +41,11 @@ use tokio::task::JoinHandle; /// terminal anyway. pub const PTY_BROADCAST_CAPACITY: usize = 256; +/// Default replay-buffer size — roughly one screenful for a typical 24×80 +/// terminal plus reasonable scrollback. Override via +/// [`PtyBridge::start_with_replay`] or the `[daemon] pty_replay_bytes` config. +pub const DEFAULT_REPLAY_BYTES: usize = 64 * 1024; + /// Maximum bytes read per syscall; chosen to match typical pipe buffer sizes /// and avoid copying huge `Vec`s through the channel. const READ_BUF_BYTES: usize = 4096; @@ -45,13 +57,20 @@ const READ_BUF_BYTES: usize = 4096; pub struct PtyBridge { fifo_path: PathBuf, sender: broadcast::Sender>, + replay: ReplayBuffer, task: JoinHandle<()>, } impl PtyBridge { /// Create the FIFO (if it doesn't exist), open it non-blocking, and start - /// the reader task. + /// the reader task with the default replay buffer size. pub async fn start(fifo_path: PathBuf) -> io::Result { + Self::start_with_replay(fifo_path, DEFAULT_REPLAY_BYTES).await + } + + /// 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 { if let Some(parent) = fifo_path.parent() { tokio::fs::create_dir_all(parent).await?; } @@ -65,13 +84,16 @@ impl PtyBridge { let async_fd = AsyncFd::with_interest(file, Interest::READABLE)?; let (sender, _) = broadcast::channel::>(PTY_BROADCAST_CAPACITY); + let replay = ReplayBuffer::new(replay_bytes); let tx = sender.clone(); + let replay_for_task = replay.clone(); let task = tokio::spawn(async move { - run_reader(async_fd, tx).await; + run_reader(async_fd, tx, replay_for_task).await; }); Ok(Self { fifo_path, sender, + replay, task, }) } @@ -80,9 +102,24 @@ impl PtyBridge { &self.fifo_path } + /// Subscribe to live PTY frames. Use [`Self::subscribe_with_replay`] for + /// new attaches that want the buffered scrollback first. pub fn subscribe(&self) -> broadcast::Receiver> { self.sender.subscribe() } + + /// Atomically snapshot the replay buffer and return a fresh subscriber + /// (NAP-229). Send the returned `Vec` to the client first; subsequent + /// live bytes arrive on the receiver. Holding the replay lock across + /// `subscribe()` keeps the reader task from publishing in between, so + /// the seam has no gap and no duplicates. + pub fn subscribe_with_replay(&self) -> (Vec, broadcast::Receiver>) { + let guard = self.replay.lock_for_snapshot(); + let snapshot = guard.snapshot(); + let rx = self.sender.subscribe(); + drop(guard); + (snapshot, rx) + } } impl Drop for PtyBridge { @@ -91,7 +128,75 @@ impl Drop for PtyBridge { } } -async fn run_reader(async_fd: AsyncFd, tx: broadcast::Sender>) { +// --------------------------------------------------------------------------- +// ReplayBuffer +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone)] +struct ReplayBuffer { + inner: Arc>, +} + +#[derive(Debug)] +struct ReplayInner { + buf: VecDeque, + cap: usize, +} + +impl ReplayBuffer { + fn new(cap: usize) -> Self { + Self { + inner: Arc::new(Mutex::new(ReplayInner { + buf: VecDeque::with_capacity(cap.min(READ_BUF_BYTES * 2)), + cap, + })), + } + } + + /// Append bytes, dropping oldest as needed to stay under `cap`. + /// Only used by tests — production calls inline this under + /// `run_reader`'s lock so the broadcast send stays atomic with the push. + #[cfg(test)] + fn push(&self, bytes: &[u8]) { + let mut s = self.inner.lock().expect("ReplayBuffer poisoned"); + if s.cap == 0 { + return; + } + // 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); + let bytes = &bytes[take_from..]; + s.buf.extend(bytes); + while s.buf.len() > s.cap { + s.buf.pop_front(); + } + } + + /// Acquire the lock for a snapshot. The caller must call `snapshot()` on + /// the returned guard before releasing the lock — that's the + /// atomicity that prevents a race with concurrent appends. + fn lock_for_snapshot(&self) -> ReplayGuard<'_> { + ReplayGuard { + inner: self.inner.lock().expect("ReplayBuffer poisoned"), + } + } +} + +struct ReplayGuard<'a> { + inner: std::sync::MutexGuard<'a, ReplayInner>, +} + +impl ReplayGuard<'_> { + fn snapshot(&self) -> Vec { + let (a, b) = self.inner.buf.as_slices(); + let mut out = Vec::with_capacity(a.len() + b.len()); + out.extend_from_slice(a); + out.extend_from_slice(b); + out + } +} + +async fn run_reader(async_fd: AsyncFd, tx: broadcast::Sender>, replay: ReplayBuffer) { let mut buf = [0u8; READ_BUF_BYTES]; loop { let mut guard = match async_fd.readable().await { @@ -113,7 +218,20 @@ async fn run_reader(async_fd: AsyncFd, tx: broadcast::Sender>) { tokio::task::yield_now().await; } Ok(Ok(n)) => { + // Hold the replay lock across both the buffer push and the + // broadcast send so `subscribe_with_replay` either sees this + // chunk in its snapshot *or* receives it on the broadcast — + // never both, never neither. + let mut s = replay.inner.lock().expect("ReplayBuffer poisoned"); + if s.cap > 0 { + let take_from = n.saturating_sub(s.cap); + s.buf.extend(&buf[take_from..n]); + while s.buf.len() > s.cap { + s.buf.pop_front(); + } + } let _ = tx.send(buf[..n].to_vec()); + drop(s); } Ok(Err(e)) => { tracing::warn!(error = ?e, "fifo read error"); @@ -154,3 +272,39 @@ fn shell_quote(s: &str) -> String { let escaped = s.replace('\'', r#"'\''"#); format!("'{escaped}'") } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn replay_buffer_keeps_last_cap_bytes() { + let r = ReplayBuffer::new(4); + r.push(b"abcdefgh"); + assert_eq!(r.lock_for_snapshot().snapshot(), b"efgh"); + } + + #[test] + fn replay_buffer_disabled_when_cap_zero() { + let r = ReplayBuffer::new(0); + r.push(b"anything"); + assert!(r.lock_for_snapshot().snapshot().is_empty()); + } + + #[test] + fn replay_buffer_handles_chunk_larger_than_cap() { + let r = ReplayBuffer::new(3); + r.push(b"abcdefghij"); + assert_eq!(r.lock_for_snapshot().snapshot(), b"hij"); + } + + #[test] + fn replay_buffer_accumulates_across_pushes() { + let r = ReplayBuffer::new(6); + r.push(b"abc"); + r.push(b"de"); + assert_eq!(r.lock_for_snapshot().snapshot(), b"abcde"); + r.push(b"fghi"); + assert_eq!(r.lock_for_snapshot().snapshot(), b"defghi"); + } +} diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 3b52111..8c85602 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -96,6 +96,9 @@ struct Inner { sidecar: SidecarStore, worktree_mgr: WorktreeManager, fifo_dir: PathBuf, + /// NAP-229: per-session PTY replay buffer cap in bytes. Each new attach + /// receives the last `replay_bytes` of output before live frames. + replay_bytes: usize, sessions: Mutex>, events: broadcast::Sender, } @@ -107,6 +110,26 @@ impl SessionManager { sidecar: SidecarStore, worktree_mgr: WorktreeManager, fifo_dir: PathBuf, + ) -> Self { + Self::new_with_replay( + daemon_id, + tmux, + sidecar, + worktree_mgr, + fifo_dir, + crate::pty::DEFAULT_REPLAY_BYTES, + ) + } + + /// Variant that lets the caller override the per-session replay buffer + /// cap. Pass `0` to disable replay (subscribers see only live frames). + pub fn new_with_replay( + daemon_id: impl Into, + tmux: Tmux, + sidecar: SidecarStore, + worktree_mgr: WorktreeManager, + fifo_dir: PathBuf, + replay_bytes: usize, ) -> Self { let (events, _) = broadcast::channel(EVENTS_CAPACITY); Self { @@ -116,6 +139,7 @@ impl SessionManager { sidecar, worktree_mgr, fifo_dir, + replay_bytes, sessions: Mutex::new(HashMap::new()), events, }), @@ -143,6 +167,17 @@ impl SessionManager { sessions.get(&id).map(|r| r.bridge.subscribe()) } + /// Subscribe to a session's PTY broadcast and return the replay buffer + /// (NAP-229). Callers send the buffer before pumping the live receiver + /// so freshly-attached clients see the last screenful immediately. + pub async fn subscribe_pty_with_replay( + &self, + id: Uuid, + ) -> Option<(Vec, broadcast::Receiver>)> { + let sessions = self.inner.sessions.lock().await; + sessions.get(&id).map(|r| r.bridge.subscribe_with_replay()) + } + pub async fn has(&self, id: Uuid) -> bool { self.inner.sessions.lock().await.contains_key(&id) } @@ -182,7 +217,8 @@ impl SessionManager { .map_err(SessionError::Tmux)?; let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); - let bridge = PtyBridge::start(fifo_path.clone()).await?; + let bridge = + PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?; self.inner .tmux .pipe_pane(id, &pipe_pane_command(&fifo_path)) @@ -297,7 +333,8 @@ impl SessionManager { }, }; let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo")); - let bridge = PtyBridge::start(fifo_path.clone()).await?; + let bridge = + PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?; // Best-effort: re-pipe so output starts flowing again. let _ = self .inner diff --git a/crates/maestro-daemon/tests/attach.rs b/crates/maestro-daemon/tests/attach.rs index 3571665..cd48eb4 100644 --- a/crates/maestro-daemon/tests/attach.rs +++ b/crates/maestro-daemon/tests/attach.rs @@ -310,3 +310,62 @@ async fn ping_text_frame_gets_pong() { assert!(saw_pong, "no Pong frame received"); let _ = h.manager.kill(id).await; } + +#[tokio::test] +async fn second_attach_sees_replay_buffer_of_prior_output() { + // NAP-229: a fresh attach must receive the buffered scrollback before + // any live frames, so reconnecting to an idle prompt isn't blank. + let h = spawn_daemon_ws("laptop").await; + let session = create_bash_session(&h).await; + let token = token_for(session, "laptop", 60); + + let mut a = connect_attach(h.addr, session, &token).await.unwrap(); + // Generate output the bash will echo back. + a.send(WsMessage::Binary(b"echo MAESTRO_REPLAY_MARK\n".to_vec())) + .await + .unwrap(); + // Drain at least until we see the mark on client A. + let want = b"MAESTRO_REPLAY_MARK"; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let mut buf_a = Vec::new(); + while std::time::Instant::now() < deadline { + tokio::select! { + msg = a.next() => match msg { + Some(Ok(WsMessage::Binary(bytes))) => buf_a.extend(bytes), + Some(Ok(_)) => {} + _ => break, + }, + _ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {} + } + if buf_a.windows(want.len()).any(|w| w == want) { + break; + } + } + assert!( + buf_a.windows(want.len()).any(|w| w == want), + "client A never saw its own echo" + ); + let _ = a.close(None).await; + + // Now attach a fresh client. It should see the prior MARK in the replay + // even though no new output has happened since. + let mut b = connect_attach(h.addr, session, &token).await.unwrap(); + let saw_replay = tokio::time::timeout(std::time::Duration::from_secs(2), async { + let mut buf = Vec::new(); + while let Some(Ok(msg)) = b.next().await { + if let WsMessage::Binary(bytes) = msg { + buf.extend(bytes); + if buf.windows(want.len()).any(|w| w == want) { + return true; + } + } + } + false + }) + .await + .unwrap_or(false); + assert!( + saw_replay, + "fresh attach did not receive replay containing prior output" + ); +}