feat(maestro): replay PTY scrollback on attach (NAP-229)
ci / rust (push) Failing after 7m57s

Each PtyBridge keeps a per-session ring buffer (Mutex<VecDeque<u8>>,
default 64 KB, override via `[daemon] pty_replay_bytes`). The reader
task appends each chunk while holding the buffer lock and broadcasts in
the same critical section, so `PtyBridge::subscribe_with_replay` —
which snapshots the buffer and subscribes to the broadcast under that
same lock — produces a snapshot that's a true prefix of subsequent live
bytes. No duplicates, no gaps at the seam.

`SessionManager` exposes the new `subscribe_pty_with_replay`, and
`attach::serve_local` sends the snapshot as a single binary frame
before pumping the live receiver. End result: attaching to an idle
bash session now shows the prompt immediately; reattaching after
running a build sees the last screenful instead of a blank terminal.

Adds `[daemon] pty_replay_bytes` to the unified config (None ⇒
default; Some(0) ⇒ disabled). E2E test exercises the full path: client
A drives output through, disconnects, client B sees the prior output
on connect.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-26 22:56:23 +00:00
co-authored by Claude Opus 4.7
parent a671e15114
commit 3bd793f394
8 changed files with 285 additions and 8 deletions
+7
View File
@@ -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<String>,
/// 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<usize>,
/// `[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()],
},
+10 -2
View File
@@ -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.
+8
View File
@@ -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<String>,
/// 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<DaemonConfig, Confi
let perimeter_id = env.perimeter_id.or(daemon.perimeter_id);
let private_addr = env.private_addr.or(daemon.private_addr);
let allow_inbound = daemon.policy.allow_inbound;
let pty_replay_bytes = daemon
.pty_replay_bytes
.unwrap_or(crate::pty::DEFAULT_REPLAY_BYTES);
Ok(DaemonConfig {
daemon_id,
@@ -211,6 +218,7 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
perimeter_id,
private_addr,
allow_inbound,
pty_replay_bytes,
})
}
+3
View File
@@ -128,6 +128,9 @@ pub fn run(
// Default deny: empty allow_inbound means no peer can attach until the
// operator opts in. Set in config.toml under [daemon.policy].
policy: Default::default(),
// None ⇒ daemon falls back to pty::DEFAULT_REPLAY_BYTES. Override in
// config.toml as `[daemon] pty_replay_bytes = ...` (NAP-229).
pty_replay_bytes: None,
};
file.save(path).map_err(ConfigError::from)?;
let _ = handle; // stdout handle only used during prompts; binary prints "wrote ..."
+2 -1
View File
@@ -121,12 +121,13 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
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?;
+157 -3
View File
@@ -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<Vec<u8>>,
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> {
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<Self> {
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::<Vec<u8>>(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<Vec<u8>> {
self.sender.subscribe()
}
/// Atomically snapshot the replay buffer and return a fresh subscriber
/// (NAP-229). Send the returned `Vec<u8>` 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<u8>, broadcast::Receiver<Vec<u8>>) {
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<File>, tx: broadcast::Sender<Vec<u8>>) {
// ---------------------------------------------------------------------------
// ReplayBuffer
// ---------------------------------------------------------------------------
#[derive(Debug, Clone)]
struct ReplayBuffer {
inner: Arc<Mutex<ReplayInner>>,
}
#[derive(Debug)]
struct ReplayInner {
buf: VecDeque<u8>,
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<u8> {
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<File>, tx: broadcast::Sender<Vec<u8>>, 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<File>, tx: broadcast::Sender<Vec<u8>>) {
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");
}
}
+39 -2
View File
@@ -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<HashMap<Uuid, SessionRecord>>,
events: broadcast::Sender<SessionEvent>,
}
@@ -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<String>,
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<u8>, broadcast::Receiver<Vec<u8>>)> {
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
+59
View File
@@ -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"
);
}