From a824f5f4d395152d8108179249ac34d73bf5e342 Mon Sep 17 00:00:00 2001 From: Edwin Date: Sat, 30 May 2026 18:52:45 -0700 Subject: [PATCH] Lazy-load older PTY replay in web UI --- crates/daemon/assets/index.html | 113 ++++++++++++++++++++++++++++++-- crates/daemon/src/server.rs | 9 ++- crates/daemon/src/session.rs | 45 ++++++++++--- crates/daemon/src/storage.rs | 70 ++++++++++++++++---- crates/protocol/src/lib.rs | 25 ++++++- 5 files changed, 228 insertions(+), 34 deletions(-) diff --git a/crates/daemon/assets/index.html b/crates/daemon/assets/index.html index d1f93e77..ca7c2c5a 100644 --- a/crates/daemon/assets/index.html +++ b/crates/daemon/assets/index.html @@ -1784,6 +1784,9 @@ const SESSION_LIST_MAX_WIDTH = 520; const WIDGET_VISIBILITY_KEY_PREFIX = "agentd.web.widgets.visible."; const WIDGET_AUTOHIDE_MS = 15000; +const PTY_REPLAY_INITIAL_BYTES = 1024 * 1024; +const PTY_REPLAY_MAX_BYTES = 8 * 1024 * 1024; +const PTY_REPLAY_TOP_THRESHOLD_ROWS = 20; let widgetAutohideTimer = null; const miniMatrixActivity = { @@ -2785,18 +2788,17 @@ replayTranscriptToTerm((t && t.events) || []); renderEditorStateForSession(id); } else { - const r = await rpc("session.pty_replay", { session_id: id }); + const r = await rpc("session.pty_replay", { + session_id: id, + max_bytes: PTY_REPLAY_INITIAL_BYTES, + }); if (r && r.size) { // Daemon's last-known size wins as the initial geometry; we // refit below in case the phone screen is narrower. state.term.resize(r.size.cols, r.size.rows); } - // Hand xterm the raw bytes (not a decoded string) so its - // built-in streaming UTF-8 decoder owns boundary handling - // across this and any subsequent `term.write` call. Passing a - // string forces a one-shot per-call decode and corrupts - // multi-byte glyphs that straddle PTY frame boundaries. - if (r && r.data) state.term.write(decodeBase64ToBytes(r.data)); + rememberPtyReplayChunk(handle, r); + await replayPtyChunks(handle); } handle.loaded = true; } catch (e) { @@ -3023,6 +3025,12 @@ loaded: false, ptyBuffer: [], ptyBuffering: false, + ptyReplayChunks: [], + ptyReplayStartOffset: 0, + ptyReplayEndOffset: 0, + ptyReplayTotalBytes: 0, + ptyReplayNextBytes: PTY_REPLAY_INITIAL_BYTES, + ptyReplayLoadingOlder: false, lastReportedSize: { cols: 0, rows: 0 }, }; state.terminalById.set(id, handle); @@ -3049,9 +3057,14 @@ host.addEventListener("wheel", (ev) => { if (state.currentId !== id) return; cancelResizeFollowBottom(); + maybeLoadOlderPtyReplay(id); ev.stopPropagation(); }, { passive: true }); + term.onScroll(() => { + if (state.currentId === id) maybeLoadOlderPtyReplay(id); + }); + installTerminalScrollContainment(host, term, id); // Tapping the terminal area focuses xterm's helper textarea, which @@ -3078,6 +3091,92 @@ return terminalHandleForSession(id) || createTerminalHandle(id); } +function rememberPtyReplayChunk(handle, result) { + if (!handle || !result || !result.data) return; + const bytes = decodeBase64ToBytes(result.data); + const start = Number(result.start_offset || 0); + const end = Number(result.end_offset || (start + bytes.length)); + if (!bytes.length && start === end) return; + if (handle.ptyReplayChunks.some((c) => c.start === start && c.end === end)) return; + handle.ptyReplayChunks.push({ start, end, bytes }); + handle.ptyReplayChunks.sort((a, b) => a.start - b.start); + handle.ptyReplayStartOffset = handle.ptyReplayChunks[0].start; + handle.ptyReplayEndOffset = handle.ptyReplayChunks[handle.ptyReplayChunks.length - 1].end; + handle.ptyReplayTotalBytes = Number(result.total_bytes || handle.ptyReplayEndOffset || 0); +} + +function writeTermBytes(term, bytes) { + if (!term || !bytes || !bytes.length) return Promise.resolve(); + return new Promise((resolve) => term.write(bytes, resolve)); +} + +async function replayPtyChunks(handle) { + if (!handle || !handle.term) return; + handle.term.reset(); + for (const chunk of handle.ptyReplayChunks) { + await writeTermBytes(handle.term, chunk.bytes); + } +} + +function ptyReplayProgressText(handle) { + const loaded = Math.max(0, (handle.ptyReplayEndOffset || 0) - (handle.ptyReplayStartOffset || 0)); + const total = handle.ptyReplayTotalBytes || loaded; + return `Loading older terminal history… ${(loaded / (1024 * 1024)).toFixed(1)} / ${(total / (1024 * 1024)).toFixed(1)} MiB`; +} + +function setTerminalLoadingMessage(text) { + terminalLoadingEl.textContent = text; + terminalLoadingEl.classList.remove("gone"); +} + +function hideTerminalLoading() { + terminalLoadingEl.classList.add("gone"); +} + +function shouldLoadOlderPtyReplay(handle) { + if (!handle || !handle.loaded || handle.ptyReplayLoadingOlder) return false; + if (handle.ptyReplayStartOffset <= 0) return false; + const active = handle.term && handle.term.buffer && handle.term.buffer.active; + if (!active) return false; + return active.viewportY <= PTY_REPLAY_TOP_THRESHOLD_ROWS; +} + +async function maybeLoadOlderPtyReplay(id) { + const handle = terminalHandleForSession(id); + if (!shouldLoadOlderPtyReplay(handle)) return; + handle.ptyReplayLoadingOlder = true; + const priorViewportY = handle.term.buffer.active.viewportY; + const priorBaseY = handle.term.buffer.active.baseY; + const maxBytes = Math.min(handle.ptyReplayNextBytes, PTY_REPLAY_MAX_BYTES); + setTerminalLoadingMessage(ptyReplayProgressText(handle)); + state.ptyBuffer = []; + state.ptyBuffering = true; + try { + const r = await rpc("session.pty_replay", { + session_id: id, + before_offset: handle.ptyReplayStartOffset, + max_bytes: maxBytes, + }); + if (state.currentId !== id) return; + rememberPtyReplayChunk(handle, r); + handle.ptyReplayNextBytes = Math.min(maxBytes * 2, PTY_REPLAY_MAX_BYTES); + setTerminalLoadingMessage(ptyReplayProgressText(handle)); + await replayPtyChunks(handle); + for (const chunk of state.ptyBuffer) await writeTermBytes(handle.term, chunk); + const newRowsAbove = Math.max(0, handle.term.buffer.active.baseY - priorBaseY); + handle.term.scrollToLine(Math.max(0, priorViewportY + newRowsAbove)); + } catch (e) { + console.warn("pty lazy replay failed:", e); + } finally { + state.ptyBuffer = []; + state.ptyBuffering = false; + handle.ptyBuffer = state.ptyBuffer; + handle.ptyBuffering = state.ptyBuffering; + handle.ptyReplayLoadingOlder = false; + hideTerminalLoading(); + } +} + if (window.visualViewport) { window.visualViewport.addEventListener("resize", refitTerminal); } diff --git a/crates/daemon/src/server.rs b/crates/daemon/src/server.rs index 0216df0e..f793a0f5 100644 --- a/crates/daemon/src/server.rs +++ b/crates/daemon/src/server.rs @@ -9,7 +9,7 @@ use agentd_protocol::{ GroupCreateResult, GroupDeleteParams, GroupMoveParams, GroupRenameParams, GroupSetCollapsedParams, Notification, PingResult, ProjectCreateParams, ProjectCreateResult, ProjectDeleteParams, ProjectDeletedNotificationPayload, ProjectMoveParams, ProjectRenameParams, - ProjectSetCollapsedParams, ProjectStateNotificationPayload, Request, Response, + ProjectSetCollapsedParams, ProjectStateNotificationPayload, PtyReplayParams, Request, Response, SessionAttachClipboardParams, SessionIdParams, SessionInputParams, SessionMoveParams, SessionPtyInputParams, SessionPtyResizeParams, SessionSetAutomodeParams, SessionSetGroupParams, SessionSetPinnedParams, SessionSetProjectParams, SessionSetTitleParams, @@ -1120,8 +1120,11 @@ async fn dispatch( } } m if m == ipc_method::SESSION_PTY_REPLAY => { - let p = params!(SessionIdParams); - match manager.pty_replay(&p.session_id).await { + let p = params!(PtyReplayParams); + match manager + .pty_replay_range(&p.session_id, p.max_bytes, p.before_offset) + .await + { Ok(r) => ok!(&r), Err(e) => Response::err(id.clone(), ErrorObject::internal(e.to_string())), } diff --git a/crates/daemon/src/session.rs b/crates/daemon/src/session.rs index 6fd5d48f..c5002e8a 100644 --- a/crates/daemon/src/session.rs +++ b/crates/daemon/src/session.rs @@ -2433,6 +2433,15 @@ impl SessionManager { } pub async fn pty_replay(&self, id: &str) -> Result { + self.pty_replay_range(id, None, None).await + } + + pub async fn pty_replay_range( + &self, + id: &str, + max_bytes: Option, + before_offset: Option, + ) -> Result { use base64::Engine; let entry = self .get_entry(id) @@ -2440,18 +2449,21 @@ impl SessionManager { .ok_or_else(|| anyhow!("session not found: {}", id))?; let size = entry.pty.lock().await.size; // Pull scrollback from the on-disk `pty.log`, not the (now-removed) - // in-memory ring. Bounded by `PTY_REPLAY_CAP`; the TUI feeds the - // bytes through a vt100 parser whose `SCROLLBACK_MAX` row budget is - // the real ceiling on what survives in scrollback. - let bytes = self + // in-memory ring. Requests are capped by `PTY_REPLAY_CAP`; clients can + // ask for older adjacent ranges and replay their local chunks in order. + let requested = max_bytes.unwrap_or(PTY_REPLAY_CAP).min(PTY_REPLAY_CAP); + let (bytes, start_offset, end_offset, total_bytes) = self .storage - .read_pty_tail(id, PTY_REPLAY_CAP) + .read_pty_range_before(id, requested, before_offset) .unwrap_or_else(|e| { - tracing::warn!(session = %id, error = ?e, "pty_log tail read failed"); - Vec::new() + tracing::warn!(session = %id, error = ?e, "pty_log range read failed"); + (Vec::new(), 0, 0, 0) }); Ok(PtyReplayResult { data: base64::engine::general_purpose::STANDARD.encode(bytes), + start_offset, + end_offset, + total_bytes, size, }) } @@ -4132,10 +4144,19 @@ mod tests { let id = "ssize"; let entry = synthetic_entry(id, agentd_protocol::SessionKind::User, 0); mgr.sessions.write().await.insert(id.into(), entry.clone()); - entry.pty.lock().await.size = Some(PtySize { cols: 132, rows: 50 }); + entry.pty.lock().await.size = Some(PtySize { + cols: 132, + rows: 50, + }); let result = mgr.pty_replay(id).await.expect("pty_replay"); - assert_eq!(result.size, Some(PtySize { cols: 132, rows: 50 })); + assert_eq!( + result.size, + Some(PtySize { + cols: 132, + rows: 50 + }) + ); } #[tokio::test] @@ -4174,7 +4195,11 @@ mod tests { let decoded = base64::engine::general_purpose::STANDARD .decode(&result.data) .expect("base64 decode"); - assert_eq!(decoded.len(), PTY_REPLAY_CAP, "replay must cap at PTY_REPLAY_CAP"); + assert_eq!( + decoded.len(), + PTY_REPLAY_CAP, + "replay must cap at PTY_REPLAY_CAP" + ); assert_eq!( decoded, bytes[extra..], diff --git a/crates/daemon/src/storage.rs b/crates/daemon/src/storage.rs index 358444b3..010e1370 100644 --- a/crates/daemon/src/storage.rs +++ b/crates/daemon/src/storage.rs @@ -512,22 +512,33 @@ impl Storage { /// a TUI's vt100 parser on attach so scrollback covers the on-disk log, /// not just a small in-memory window. pub fn read_pty_tail(&self, id: &str, max_bytes: usize) -> Result> { + let (bytes, _, _, _) = self.read_pty_range_before(id, max_bytes, None)?; + Ok(bytes) + } + + /// Read up to `max_bytes` ending before `before_offset` from `pty.log`. + /// Offsets are absolute byte offsets in the file; `before_offset: None` + /// means the current end. Returns `(bytes, start, end, total_len)`. + pub fn read_pty_range_before( + &self, + id: &str, + max_bytes: usize, + before_offset: Option, + ) -> Result<(Vec, u64, u64, u64)> { let path = self.pty_log_path(id); - if !path.exists() { - return Ok(Vec::new()); + if !path.exists() || max_bytes == 0 { + return Ok((Vec::new(), 0, 0, 0)); } let mut f = std::fs::File::open(&path).with_context(|| format!("open {}", path.display()))?; - let len = f.metadata()?.len() as usize; - let offset = len.saturating_sub(max_bytes); - if offset > 0 { - use std::io::Seek; - f.seek(std::io::SeekFrom::Start(offset as u64))?; - } - use std::io::Read; - let mut buf = Vec::with_capacity(len - offset); - f.read_to_end(&mut buf)?; - Ok(buf) + let total = f.metadata()?.len(); + let end = before_offset.unwrap_or(total).min(total); + let start = end.saturating_sub(max_bytes as u64); + use std::io::{Read, Seek}; + f.seek(std::io::SeekFrom::Start(start))?; + let mut buf = vec![0; (end - start) as usize]; + f.read_exact(&mut buf)?; + Ok((buf, start, end, total)) } /// Remove the entire session directory (meta + transcript + worktree). @@ -751,6 +762,41 @@ mod transcript_tail_tests { } } +#[cfg(test)] +mod pty_range_tests { + use super::*; + + #[test] + fn pty_range_before_returns_offsets_and_requested_slice() { + let tmp = tempfile::tempdir().unwrap(); + let storage = Storage::new(tmp.path().join("data")).unwrap(); + storage + .append_pty_bytes("s1", b"abcdefghijklmnopqrstuvwxyz") + .unwrap(); + + let (bytes, start, end, total) = storage.read_pty_range_before("s1", 5, Some(20)).unwrap(); + + assert_eq!(bytes, b"pqrst"); + assert_eq!(start, 15); + assert_eq!(end, 20); + assert_eq!(total, 26); + } + + #[test] + fn pty_range_before_clamps_to_file_start_and_end() { + let tmp = tempfile::tempdir().unwrap(); + let storage = Storage::new(tmp.path().join("data")).unwrap(); + storage.append_pty_bytes("s1", b"abcdef").unwrap(); + + let (bytes, start, end, total) = storage.read_pty_range_before("s1", 99, Some(99)).unwrap(); + + assert_eq!(bytes, b"abcdef"); + assert_eq!(start, 0); + assert_eq!(end, 6); + assert_eq!(total, 6); + } +} + #[cfg(test)] mod memory_tests { use super::*; diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 5a352a6f..d47954c5 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -980,14 +980,35 @@ pub struct LoopRemoveParams { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PtyReplayResult { - /// Base64-encoded raw bytes representing the recent PTY history (best - /// effort — the daemon keeps a bounded ring buffer). + /// Base64-encoded raw bytes representing the requested PTY log range. pub data: String, + /// Absolute byte offset where `data` starts in `pty.log`. + #[serde(default)] + pub start_offset: u64, + /// Absolute byte offset where `data` ends in `pty.log`. + #[serde(default)] + pub end_offset: u64, + /// Current total byte length of `pty.log`. + #[serde(default)] + pub total_bytes: u64, /// Most recent known PTY size for the session, if available. #[serde(default, skip_serializing_if = "Option::is_none")] pub size: Option, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PtyReplayParams { + pub session_id: String, + /// Return up to this many bytes before `before_offset`. When omitted, + /// defaults to the historical replay cap from the current log tail. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_bytes: Option, + /// Exclusive end offset for the requested range. When omitted, uses the + /// current end of `pty.log`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub before_offset: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SessionDetail { pub summary: SessionSummary,