diff --git a/Cargo.lock b/Cargo.lock index 69958628352..7a804448a20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2657,6 +2657,7 @@ dependencies = [ "tempfile", "tokio", "toml 0.9.12+spec-1.1.0", + "tracing", "unicode-width 0.2.0", "uuid", ] diff --git a/crates/fractal-api-client-v0/src/api/client_v0/raw_terminal.rs b/crates/fractal-api-client-v0/src/api/client_v0/raw_terminal.rs index e05bd0c9675..fe5bd64b07b 100644 --- a/crates/fractal-api-client-v0/src/api/client_v0/raw_terminal.rs +++ b/crates/fractal-api-client-v0/src/api/client_v0/raw_terminal.rs @@ -134,8 +134,18 @@ pub async fn stream( Query(query): Query, headers: HeaderMap, ) -> Result { + let started = std::time::Instant::now(); + tracing::debug!( + terminal = %id, + incarnation = %query.incarnation, + mode = mode_name(query.mode), + started_at_unix_ms = %client_now_ms(), + phase = "auth", + "raw terminal setup started" + ); let upgrade = upgrade.and_then(|Extension(context)| context.0); authorize(&session, query.mode)?; + tracing::debug!(terminal = %id, phase = "auth", elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase completed"); // This protocol subscribes in the upgrade request, without a JSON subscribe command. let first_frame = subscription_first_frame("terminal", "raw", upgrade.as_ref(), None) .map(|first| first.span); @@ -152,10 +162,14 @@ pub async fn stream( &protocols, SUBPROTOCOL, query.mode == fractal_client::RawTerminalMode::Peek, ).ok_or_else(|| validation("raw terminal requires one supported pty protocol plus one capability protocol (optional for peek)"))?; let direct_peek = capability.is_none(); + let lookup_started = std::time::Instant::now(); + tracing::debug!(terminal = %id, phase = "lookup", elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase started"); let live = remote_terminal_live_session(&state, &terminal_subject(&id), &query.incarnation)?; if !live.terminal { return Err(validation("raw attachment requires a terminal runtime")); } + tracing::debug!(terminal = %id, owner = %live.owner_host_id, phase = "lookup", duration_ms = %lookup_started.elapsed().as_millis(), elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase completed"); + let authorization_started = std::time::Instant::now(); let acquisition_epoch = if direct_peek { Some(authorization_epoch(&state, &session)?) } else if query.mode == fractal_client::RawTerminalMode::Peek { @@ -199,6 +213,9 @@ pub async fn stream( ) }) .transpose()?; + tracing::debug!(terminal = %id, phase = "capability", duration_ms = %authorization_started.elapsed().as_millis(), elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase completed"); + let open_started = std::time::Instant::now(); + tracing::debug!(terminal = %id, owner = %live.owner_host_id, phase = "owner-open", elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase started"); // Open and fence before HTTP upgrade: a stale owner incarnation is a refusal, not a blank pane. let (transport, control) = if live.owner_host_id == client_host_id(&state.node) { let terminal = crate::model::LocalTerminal { @@ -232,6 +249,7 @@ pub async fn stream( .map_err(|error| remote_read_error(&live.owner_host_id, error))?; (transport, Some(control)) }; + tracing::debug!(terminal = %id, owner = %live.owner_host_id, phase = "owner-open", duration_ms = %open_started.elapsed().as_millis(), elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase completed"); if let Some(lease) = lease.clone() { lease_query(lease, |lease| lease.revalidate()) .await @@ -247,11 +265,25 @@ pub async fn stream( "raw PEEK open authorized" ); } + tracing::debug!(terminal = %id, phase = "upgrade", elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase started"); Ok(websocket .protocols([selected.to_owned()]) .max_message_size(CHUNK * 4) .max_frame_size(CHUNK * 4) - .on_upgrade(move |socket| splice(socket, transport, Some(query.mode), lease, control, upgrade, first_frame))) + .on_upgrade(move |socket| async move { + tracing::debug!(terminal = %id, phase = "upgrade", elapsed_ms = %started.elapsed().as_millis(), "raw terminal phase completed"); + let span = tracing::debug_span!( + "raw_terminal_stream", + terminal = %id, + incarnation = %query.incarnation, + mode = mode_name(query.mode) + ); + tracing::Instrument::instrument( + splice(socket, transport, Some(query.mode), lease, control, upgrade, first_frame), + span, + ) + .await; + })) } #[cfg(test)] diff --git a/crates/fractal-api-core/src/api/client_v0/raw_terminal.rs b/crates/fractal-api-core/src/api/client_v0/raw_terminal.rs index 2e7b6bfa292..d75a50e64b1 100644 --- a/crates/fractal-api-core/src/api/client_v0/raw_terminal.rs +++ b/crates/fractal-api-core/src/api/client_v0/raw_terminal.rs @@ -34,6 +34,8 @@ pub async fn splice( _upgrade: Option, mut first_frame: Option, ) { + let started = std::time::Instant::now(); + let mut first_frame_sent = false; let (mut sink, mut source) = socket.split(); let (mut reader, mut writer) = transport.into_split(); let proof_ready = tokio::sync::Notify::new(); @@ -144,6 +146,15 @@ pub async fn splice( } let first_span = first_frame.clone(); if first_frame_work(first_span, sink.send(axum::extract::ws::Message::Binary(bytes[..count].to_vec().into()))).await.is_err() { break; } + if !first_frame_sent { + tracing::debug!( + phase = "first-frame", + bytes = count, + elapsed_ms = %started.elapsed().as_millis(), + "raw terminal first frame sent after upgrade" + ); + first_frame_sent = true; + } first_frame = None; } } diff --git a/crates/fractal-peer/src/peer.rs b/crates/fractal-peer/src/peer.rs index 6ed7dd9e02d..2c994353913 100644 --- a/crates/fractal-peer/src/peer.rs +++ b/crates/fractal-peer/src/peer.rs @@ -1190,6 +1190,8 @@ async fn receive_raw_terminal( axum::Extension(splice): axum::Extension, headers: HeaderMap, ) -> Response { + let started = std::time::Instant::now(); + tracing::debug!(terminal = %query.terminal, incarnation = %query.incarnation, phase = "peer-auth", "raw terminal owner setup started"); let profile = crate::profile::Op::start("GET /v1/peer/raw-terminal", None); let path = uri.path_and_query().map_or(uri.path(), |value| value.as_str()); if path.len() > 16_384 || !query.person.starts_with("person/") || query.person.matches('/').count() != 1 { @@ -1201,6 +1203,7 @@ async fn receive_raw_terminal( _ => return (StatusCode::UNAUTHORIZED, "untrusted raw terminal member").into_response(), }; drop(auth_span); + tracing::debug!(terminal = %query.terminal, phase = "peer-auth", elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase completed"); if query.mode == fractal_client::RawTerminalMode::Peek { let binding: RawLeaseBinding = match query .binding @@ -1241,14 +1244,20 @@ async fn receive_raw_terminal( let client = fractal_client::Client::unix_as(state.backend().socket(), &query.person); let transport = async { let attachment_span = profile.as_ref().map(|op| op.wall_span("raw/owner-attachment")); + let attachment_started = std::time::Instant::now(); + tracing::debug!(terminal = %query.terminal, phase = "owner-attachment", elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase started"); let attachment = client.raw_terminal_attachment(&query.terminal, &query.incarnation, query.mode).await?; drop(attachment_span); + tracing::debug!(terminal = %query.terminal, phase = "owner-attachment", duration_ms = %attachment_started.elapsed().as_millis(), elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase completed"); if attachment.owner_host_id != format!("host/{}", state.node()) { return Err(fractal_client::ClientError::Protocol("raw terminal route is not owner-local".into())); } let open_span = profile.as_ref().map(|op| op.wall_span("raw/owner-open")); + let open_started = std::time::Instant::now(); + tracing::debug!(terminal = %query.terminal, phase = "owner-open", elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase started"); let transport = client.raw_terminal_stream(&attachment).await; drop(open_span); + tracing::debug!(terminal = %query.terminal, phase = "owner-open", success = transport.is_ok(), duration_ms = %open_started.elapsed().as_millis(), elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase completed"); transport }.await; if let Some(profile) = profile { @@ -1264,7 +1273,10 @@ async fn receive_raw_terminal( Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, "raw terminal signature failed").into_response(), }; let mut response = websocket.max_message_size(64 * 1024).max_frame_size(64 * 1024) - .on_upgrade(move |socket| splice(socket, transport)); + .on_upgrade(move |socket| async move { + tracing::debug!(terminal = %query.terminal, phase = "peer-upgrade", elapsed_ms = %started.elapsed().as_millis(), "raw terminal owner phase completed"); + splice(socket, transport).await; + }); response.headers_mut().extend(signed); response } diff --git a/crates/fractal-tui/Cargo.toml b/crates/fractal-tui/Cargo.toml index 4c6244a51a7..e6757449944 100644 --- a/crates/fractal-tui/Cargo.toml +++ b/crates/fractal-tui/Cargo.toml @@ -20,6 +20,7 @@ fractal-conversation-ui = { path = "../fractal-conversation-ui", features = ["ra fractal-mission-ui = { path = "../fractal-mission-ui", features = ["ratatui"] } fractal-ui-model = { path = "../fractal-ui-model" } tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } +tracing = "0.1" uuid = { version = "1", features = ["v7"] } signal-hook = "0.3" serde_json = "1" diff --git a/crates/fractal-tui/src/ui/attach_trace.rs b/crates/fractal-tui/src/ui/attach_trace.rs new file mode 100644 index 00000000000..a45bcb82825 --- /dev/null +++ b/crates/fractal-tui/src/ui/attach_trace.rs @@ -0,0 +1,216 @@ +//! Opt-in, bounded attach diagnostics. Never records capabilities, payloads or error text. +use std::{collections::{BTreeMap, BTreeSet}, fs::{File, OpenOptions}, io::Write as _, os::{fd::AsRawFd as _, unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _}}, sync::{Arc, Mutex}, time::Instant}; + +const MAX_BYTES: u64 = 4 * 1024 * 1024; + +#[derive(Clone)] +pub(crate) struct Trace(Arc); + +struct Inner { + attempt: uuid::Uuid, + agent: String, + started: Instant, + state: Mutex, +} + +#[derive(Default)] +struct State { + finished: bool, + next: u64, + last_phase: Option<&'static str>, + active: BTreeMap, + milestones: BTreeSet<&'static str>, + sink: Option, + #[cfg(test)] + captured: Option>>>, + #[cfg(test)] + info_outcomes: usize, +} + +impl Trace { + pub(crate) fn new(agent: &str) -> Self { + let sink = std::env::var_os("FRACTAL_TUI_ATTACH_TRACE_FILE").and_then(|path| { + let path = std::path::Path::new(&path); + if !path.is_absolute() { return None; } + let file = OpenOptions::new().append(true).create(true).mode(0o600) + .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK).open(path).ok()?; + let metadata = file.metadata().ok()?; + // Refuse non-files, other owners, hardlinks, and existing permissive files. + if !metadata.is_file() || metadata.permissions().mode() & 0o777 != 0o600 + || metadata.uid() != unsafe { libc::geteuid() } || metadata.nlink() != 1 { return None; } + Some(file) + }); + Self::with_state(agent, State { sink, ..State::default() }) + } + + fn with_state(agent: &str, state: State) -> Self { + let trace = Self(Arc::new(Inner { attempt: uuid::Uuid::now_v7(), agent: agent.to_owned(), started: Instant::now(), state: Mutex::new(state) })); + trace.0.emit(&mut trace.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner), "attempt", "start", None); + trace + } + + pub(crate) fn milestone(&self, phase: &'static str) { + let mut state = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if !state.finished && state.milestones.insert(phase) { + state.last_phase = Some(phase); + self.0.emit(&mut state, phase, "success", None); + } + } + + pub(crate) fn complete(&self) { + self.milestone("first-screen-rendered"); + self.finish("success"); + } + + pub(crate) fn finish(&self, outcome: &'static str) { + self.0.finish(outcome); + } + + pub(crate) fn start_phase(&self, phase: &'static str) -> Phase { + let mut state = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + let id = state.next; + state.next += 1; + if !state.finished { + state.last_phase = Some(phase); + state.active.insert(id, (phase, Instant::now())); + self.0.emit(&mut state, phase, "start", None); + } + Phase { trace: self.clone(), id } + } + + #[cfg(test)] + pub(crate) fn captured(agent: &str) -> (Self, Arc>>) { + let events = Arc::new(Mutex::new(Vec::new())); + (Self::with_state(agent, State { captured: Some(events.clone()), ..State::default() }), events) + } +} + +impl Inner { + fn emit(&self, state: &mut State, phase: &'static str, outcome: &'static str, duration: Option) { + #[cfg(not(test))] + if state.sink.is_none() { return; } + #[cfg(test)] + if state.sink.is_none() && state.captured.is_none() { return; } + let level = if matches!(outcome, "start" | "success") { "DEBUG" } else { "INFO" }; + let at_unix_ms = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_millis()); + let event = serde_json::json!({"service":"fractal-tui", "level":level, "attempt_id":self.attempt.to_string(), "agent":self.agent, "phase":phase, "phase_reached":state.last_phase, "outcome":outcome, "at_unix_ms":at_unix_ms, "elapsed_ms":self.started.elapsed().as_millis(), "phase_elapsed_ms":duration}); + #[cfg(test)] + if let Some(events) = &state.captured { events.lock().unwrap().push(event.clone()); } + if let Some(file) = state.sink.as_mut() { + if let Ok(mut bytes) = serde_json::to_vec(&event) { + bytes.push(b'\n'); + // Serialize append + size check without waiting behind another process: + // diagnostics must not introduce another attach stall. + // SAFETY: flock borrows the live file descriptor and retains no pointers. + if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0 { + if file.metadata().is_ok_and(|metadata| metadata.len().saturating_add(bytes.len() as u64) <= MAX_BYTES) { + let _ = file.write_all(&bytes); + } + // SAFETY: the descriptor still belongs to this live File. + let _ = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_UN) }; + } + } + } + } + + fn finish(&self, outcome: &'static str) { + let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if state.finished { return; } + state.finished = true; + if outcome != "success" { + tracing::info!( + target: "fractal_tui::attach", + attempt_id = %self.attempt, + agent = %self.agent, + phase_reached = state.last_phase.unwrap_or("attempt"), + outcome, + elapsed_ms = %self.started.elapsed().as_millis(), + "terminal attach attempt ended" + ); + // Unit tests capture this outcome in memory rather than touch the user's cache. + #[cfg(not(test))] + super::lastrun::note_attach_outcome( + self.attempt, &self.agent, state.last_phase.unwrap_or("attempt"), + outcome, self.started.elapsed().as_millis(), + ); + #[cfg(test)] + { state.info_outcomes += 1; } + } + for (_, (phase, started)) in std::mem::take(&mut state.active) { + self.emit(&mut state, phase, outcome, Some(started.elapsed().as_millis())); + } + self.emit(&mut state, "attempt", outcome, None); + } +} + +impl Drop for Inner { + fn drop(&mut self) { self.finish("dropped"); } +} + +pub(crate) struct Phase { trace: Trace, id: u64 } + +impl Phase { + pub(crate) fn end(self, outcome: &'static str) { self.record(outcome); } + fn record(&self, outcome: &'static str) { + let mut state = self.trace.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some((phase, started)) = state.active.remove(&self.id) { + self.trace.0.emit(&mut state, phase, outcome, Some(started.elapsed().as_millis())); + } + } +} + +impl Drop for Phase { + fn drop(&mut self) { self.record("cancelled"); } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn interleaved_attempts_are_isolated_and_terminal_once() { + let (a, first) = Trace::captured("agent/a"); + let (b, second) = Trace::captured("agent/b"); + let pa = a.start_phase("upgrade"); + let pb = b.start_phase("upgrade"); + a.finish("selection-changed"); + drop(pa); + pb.end("success"); + b.milestone("first-frame-decoded"); + b.complete(); + b.finish("dropped"); + let first = first.lock().unwrap(); + let second = second.lock().unwrap(); + assert_ne!(first[0]["attempt_id"], second[0]["attempt_id"]); + for (events, agent, outcome) in [(&*first, "agent/a", "selection-changed"), (&*second, "agent/b", "success")] { + assert!(events.iter().all(|event| event["agent"] == agent && event["attempt_id"] == events[0]["attempt_id"])); + let terminal: Vec<_> = events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] != "start").collect(); + assert_eq!(terminal.len(), 1); + assert_eq!(terminal[0]["outcome"], outcome); + assert_eq!(terminal[0]["phase_reached"], if agent == "agent/a" { "upgrade" } else { "first-screen-rendered" }); + let trace = if agent == "agent/a" { &a } else { &b }; + assert_eq!(trace.0.state.lock().unwrap().info_outcomes, usize::from(outcome != "success")); + assert!(terminal[0]["at_unix_ms"].as_u64().is_some_and(|time| time > 0)); + } + } + + #[test] + fn final_owner_drop_records_terminal_outcome() { + let (trace, events) = Trace::captured("agent/a"); + drop(trace); + assert_eq!(events.lock().unwrap().last().unwrap()["outcome"], "dropped"); + } + #[test] + fn unsuccessful_terminal_outcomes_log_once_without_an_opt_in_sink() { + for outcome in ["outer-deadline", "selection-changed", "dropped", "setup-failed"] { + let trace = Trace::with_state("agent/a", State::default()); + trace.finish(outcome); + trace.finish(outcome); + assert_eq!(trace.0.state.lock().unwrap().info_outcomes, 1); + } + let trace = Trace::with_state("agent/a", State::default()); + trace.complete(); + assert_eq!(trace.0.state.lock().unwrap().info_outcomes, 0); + } +} diff --git a/crates/fractal-tui/src/ui/context.rs b/crates/fractal-tui/src/ui/context.rs index 40d205b856e..a209e48a80d 100644 --- a/crates/fractal-tui/src/ui/context.rs +++ b/crates/fractal-tui/src/ui/context.rs @@ -860,6 +860,7 @@ mod tests { "one".into(), 24, 80, + None, )), }); peer.write_all(&encode_packet( diff --git a/crates/fractal-tui/src/ui/glass.rs b/crates/fractal-tui/src/ui/glass.rs index 2ce646012cb..90db0ecdeb1 100644 --- a/crates/fractal-tui/src/ui/glass.rs +++ b/crates/fractal-tui/src/ui/glass.rs @@ -5416,6 +5416,7 @@ mod tests { let (client, mut peer) = UnixStream::pair().unwrap(); ui.terminal.as_mut().unwrap().native = Some(pty::NativeTerminal::spawn( client, "agent/example/atlas/builder", "one".into(), 24, 80, + None, )); peer.write_all(&encode_packet(MessageType::Screen, b"\x1b[?1000h\x1b[?1006h")).unwrap(); let start = Instant::now(); @@ -5531,6 +5532,7 @@ mod tests { let (client, _peer) = UnixStream::pair().unwrap(); ui.terminal.as_mut().unwrap().native = Some(pty::NativeTerminal::spawn( client, "agent/example/atlas/builder", "one".into(), 24, 80, + None, )); let mut terminal = Terminal::new(TestBackend::new(width, 24)).unwrap(); terminal.draw(|frame| ui.render(frame)).unwrap(); @@ -5558,6 +5560,7 @@ mod tests { peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); ui.terminal.as_mut().unwrap().native = Some(pty::NativeTerminal::spawn( client, "agent/example/atlas/builder", "one".into(), 24, 200, + None, )); peer.write_all(&encode_packet( MessageType::Screen, b"\x1b[?1003h\x1b[?1006h\x1b[?1004h", @@ -8048,7 +8051,7 @@ mod tests { let (client, mut peer) = UnixStream::pair().unwrap(); peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); ui.terminal_view_mut(agent).unwrap().native = Some( - crate::ui::pty::NativeTerminal::spawn(client, agent, "one".into(), 24, 80), + crate::ui::pty::NativeTerminal::spawn(client, agent, "one".into(), 24, 80, None), ); peer.write_all(&encode_packet( MessageType::Screen, @@ -10206,6 +10209,7 @@ mod tests { "one".into(), 24, 80, + None, )), }); let mut reader = PacketReader::new(); @@ -10299,6 +10303,7 @@ mod tests { "one".into(), 24, 80, + None, )), }); let mut reader = PacketReader::new(); diff --git a/crates/fractal-tui/src/ui/lastrun.rs b/crates/fractal-tui/src/ui/lastrun.rs index aa6f0849029..c5c7006fb25 100644 --- a/crates/fractal-tui/src/ui/lastrun.rs +++ b/crates/fractal-tui/src/ui/lastrun.rs @@ -150,6 +150,19 @@ pub fn note_slow(what: &str, context: &str, build: &str) { } } +/// Retain non-successful attach outcomes even when no tracing subscriber is installed. +#[cfg(not(test))] +pub(crate) fn note_attach_outcome(attempt_id: uuid::Uuid, agent: &str, phase_reached: &str, outcome: &str, elapsed_ms: u128) { + if let Some(file) = directory().and_then(|runs| runs.parent().map(|dir| dir.join("attach.log"))) { + let event = serde_json::json!({ + "service": "fractal-tui", "level": "INFO", "at_unix_ms": now_unix() * 1000, + "attempt_id": attempt_id.to_string(), "agent": agent, + "phase_reached": phase_reached, "outcome": outcome, "elapsed_ms": elapsed_ms, + }); + write_line(&file, &event.to_string()); + } +} + /// Append one line to a log in the cache that is cleared once it passes 100 KB. fn write_line(file: &Path, line: &str) { use std::io::Write as _; diff --git a/crates/fractal-tui/src/ui/live.rs b/crates/fractal-tui/src/ui/live.rs index fc88a98fa3c..250db96031c 100644 --- a/crates/fractal-tui/src/ui/live.rs +++ b/crates/fractal-tui/src/ui/live.rs @@ -1548,6 +1548,7 @@ pub fn run(context: Context) -> Result<()> { } Effect::OpenTerminal { agent } => { let (generation, wanted, serial) = attach.begin(&mut ui, &agent); + let trace = attach.trace(&agent); // The PTY session's own bytes, through st's raw stream to whichever host owns // it; st's screen view (the feed follows it on its socket) only when st cannot // give a direct stream. @@ -1590,19 +1591,22 @@ pub fn run(context: Context) -> Result<()> { let task_agent = agent.clone(); let agent = agent.clone(); let fallback = ui.terminal_fallback; + let setup_owner = attach::SetupOwner::new(trace.clone()); let task = runtime.spawn(async move { - // A shell has no other view to fall back to, and one started a moment ago - // can be refused while st settles which host runs it: wait, say why, retry. + let mut setup_owner = setup_owner; + // Discovery refusals may settle for a new shell. Capability timeouts + // retry inside attach_terminal, keeping its selected incarnation. let delays: &[Duration] = if agent.starts_with("terminal/") { &attach::RETRY_DELAYS } else { &[] }; - let attached = attach::phase("attach setup", attach::DEADLINE, async { + let attached = attach::traced_phase(trace.as_ref(), "attach setup", attach::DEADLINE, async { let _permit = serial.acquire_owned().await .map_err(|_| "attach setup cancelled".to_owned())?; retry_attach( delays, + true, || wanted.load(Ordering::Relaxed), |reason| { let _ = tx.send(Fetched::AttachWaiting { @@ -1619,10 +1623,16 @@ pub fn run(context: Context) -> Result<()> { &runtime_ids, owner.as_deref(), fallback, + trace.as_ref(), ) }, ).await }).await; + if let Err(reason) = &attached { + if let Some(trace) = &trace { + trace.finish(if reason == "attach setup timed out" { "outer-deadline" } else { "setup-failed" }); + } + } // Its tab closed while it waited: nobody is left to hear the outcome. if !wanted.load(Ordering::Relaxed) { if let Ok(direct) = attached { attach::close(direct); } @@ -1636,8 +1646,11 @@ pub fn run(context: Context) -> Result<()> { runtime_ids, reason, }, - }) && let Fetched::Native { direct, .. } = error.0 { - attach::close(direct); + }) { + if let Some(trace) = &trace { trace.finish("dispatch-unavailable"); } + if let Fetched::Native { direct, .. } = error.0 { attach::close(direct); } + } else { + setup_owner.handed_off(); } }); attach.tasks.insert(task_agent, task); @@ -1965,6 +1978,9 @@ pub fn run(context: Context) -> Result<()> { attach.opens.retain(|target, _| ui.terminal_view(target).is_some()); ui.step_drafts(); execute!(io::stdout(), BeginSynchronizedUpdate)?; + for native in ui.terminal.iter().chain(ui.parked.iter()).filter_map(|view| view.native.as_ref()) { + native.frame_started(); + } let mut links = Vec::new(); let draw_started = Instant::now(); terminal.draw(|frame| { @@ -2033,6 +2049,9 @@ pub fn run(context: Context) -> Result<()> { || execute!(io::stdout(), EndSynchronizedUpdate), |target| ui.conversation_entry_visible(target), )?; + for native in ui.terminal.iter().chain(ui.parked.iter()).filter_map(|view| view.native.as_ref()) { + native.frame_rendered(); + } ui.quality.live(extras.live); let top = u8::from(extras.live) | (u8::from(extras.offline.is_some()) << 1) @@ -2115,6 +2134,7 @@ pub fn run(context: Context) -> Result<()> { Some(&expected), owner.as_deref(), fallback, + None, ) .await; let _ = tx.send(Fetched::Reattached { agent, outcome }); @@ -3239,6 +3259,7 @@ mod tests { .unwrap() .block_on(retry_attach( &vec![Duration::ZERO; delays], + true, || true, |reason| said.push(reason.to_owned()), || { @@ -3284,11 +3305,58 @@ mod tests { Some(&fractal_client::ErrorCode::NotFound), "terminal/x" ))); - // Transient shell discovery may retry briefly, but a phase timeout is never retried. + // Capability issuance opens no PTY; upgrade cancellation has no cleanup receipt. assert_eq!(super::attach::RETRY_DELAYS.iter().sum::(), Duration::from_secs(3)); + assert!(super::worth_retrying("raw terminal capability timed out")); + assert!(!super::worth_retrying("raw terminal connect/upgrade (including auth and owner-open) timed out")); + assert!(!super::worth_retrying("raw terminal first-frame timed out")); assert!(!super::worth_retrying("raw terminal stream timed out")); } + #[test] + fn setup_timeouts_retry_only_capability_and_drop_each_attempt_before_retry() { + use std::sync::{Arc, atomic::{AtomicUsize, Ordering}}; + struct Cleanup(Arc); + impl Drop for Cleanup { + fn drop(&mut self) { self.0.fetch_add(1, Ordering::SeqCst); } + } + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time().build().unwrap(); + for phase in [ + attach::SetupPhase::Capability, + attach::SetupPhase::Upgrade, + attach::SetupPhase::FirstFrame, + ] { + let attempts = Arc::new(AtomicUsize::new(0)); + let cleaned = Arc::new(AtomicUsize::new(0)); + let result: Result<(), String> = runtime.block_on(retry_attach( + &[Duration::ZERO; attach::RETRY_DELAYS.len()], + false, + || true, + |_| {}, + || { + let attempt = attempts.fetch_add(1, Ordering::SeqCst); + assert_eq!(cleaned.load(Ordering::SeqCst), attempt, + "the previous setup future must be dropped before another starts"); + let guard = Cleanup(cleaned.clone()); + async move { + attach::phase(phase.name(), Duration::from_millis(1), async move { + let _guard = guard; + std::future::pending::>().await + }).await + } + }, + )); + let reason = result.unwrap_err(); + assert_eq!(attach::SetupPhase::timeout(&reason), Some(phase)); + let expected = if phase.retry_safe() { 3 } else { 1 }; + assert_eq!(attempts.load(Ordering::SeqCst), expected); + assert_eq!(cleaned.load(Ordering::SeqCst), expected); + } + assert_eq!(attach::SetupPhase::timeout("raw terminal stream timed out"), None); + assert_eq!(attach::SetupPhase::timeout("raw terminal capability refused"), None); + } + #[test] fn a_shells_attach_stops_retrying_once_its_tab_is_closed() { let unsure = "Fractal cannot tell yet which host runs this"; @@ -3301,6 +3369,7 @@ mod tests { .unwrap() .block_on(retry_attach( &[Duration::ZERO; 20], + true, || wanted.get(), |_| { said += 1; @@ -5080,6 +5149,9 @@ fn connecting_text(waited: Duration) -> String { /// runs the terminal, that host is briefly out of reach, or st asked for a pause. A terminal that /// is gone, restarted or not allowed is not retried. fn worth_retrying(reason: &str) -> bool { + if let Some(phase) = attach::SetupPhase::timeout(reason) { + return phase.retry_safe(); + } [ "cannot tell yet which host", "cannot be reached right now", @@ -5095,6 +5167,7 @@ fn worth_retrying(reason: &str) -> bool { /// as `wanted` says nobody is waiting any more (the tab was closed). async fn retry_attach( delays: &[Duration], + transient_refusals: bool, wanted: impl Fn() -> bool, mut waiting: impl FnMut(&str), mut attempt: impl FnMut() -> Fut, @@ -5105,7 +5178,8 @@ where let mut delays = delays.iter(); loop { match attempt().await { - Err(reason) if worth_retrying(&reason) => match delays.next() { + Err(reason) if worth_retrying(&reason) + && (transient_refusals != attach::SetupPhase::timeout(&reason).is_some()) => match delays.next() { Some(delay) if wanted() => { waiting(&reason); tokio::time::sleep(*delay).await; @@ -5128,6 +5202,7 @@ async fn attach_known_then_direct( runtime_ids: &[String], owner: Option<&str>, fallback: bool, + trace: Option<&super::attach_trace::Trace>, ) -> Result { if let Some((name, terminal, incarnation)) = known && !agent.starts_with("terminal/") @@ -5135,10 +5210,10 @@ async fn attach_known_then_direct( // Once selected, keep this incarnation through every route failure. Rediscovery after // an error could silently attach a process that restarted in the meantime. return attach_terminal( - client, agent, (name, incarnation), &terminal, owner, None, fallback, + client, agent, (name, incarnation), &terminal, owner, None, fallback, trace, ).await; } - attach_direct(client, agent, runtime_ids, None, owner, fallback).await + attach_direct(client, agent, runtime_ids, None, owner, fallback, trace).await } async fn attach_direct( @@ -5148,8 +5223,9 @@ async fn attach_direct( expected: Option<&str>, owner: Option<&str>, fallback: bool, + trace: Option<&super::attach_trace::Trace>, ) -> Result { - let found = attach::phase("terminal discovery", attach::DISCOVERY, async { + let found = attach::traced_phase(trace, "terminal discovery", attach::DISCOVERY, async { let mut found = Vec::new(); if subject.starts_with("terminal/") { // A shell just started may take a moment before its screen exists. @@ -5207,6 +5283,7 @@ async fn attach_direct( owner.as_deref(), expected, fallback, + trace, ) .await { @@ -5229,9 +5306,10 @@ async fn attach_terminal( owner: Option<&str>, expected: Option<&str>, fallback: bool, + trace: Option<&super::attach_trace::Trace>, ) -> Result { let (name, incarnation) = identity; - let opened = attach::phase("direct open", attach::DIRECT, direct::open( + let opened = attach::traced_phase(trace, "direct open", attach::DIRECT, direct::open( client, &direct::Terminal { subject, @@ -5259,13 +5337,21 @@ async fn attach_terminal( } direct::Opened::UseGateway { route } => route, }; - let attachment = attach::phase("raw terminal attachment", attach::GATEWAY, async { - client.raw_terminal_attachment(terminal, &incarnation, fractal_client::RawTerminalMode::Attach) - .await.map_err(|error| format!("{route}: {}", error.human())) - }).await?; + // Retry only this pre-open phase, not discovery: every attempt keeps the same incarnation. + let attachment = retry_attach( + &attach::RETRY_DELAYS, + false, + || true, + |_| {}, + || attach::traced_phase(trace, attach::SetupPhase::Capability.name(), attach::GATEWAY, async { + client.raw_terminal_attachment(terminal, &incarnation, fractal_client::RawTerminalMode::Attach) + .await.map_err(|error| format!("{route}: {}", error.human())) + }), + ).await?; // An unused capability opens no PTY and expires on the server after 60 seconds. - // Once upgraded, dropping the stream cancels its bridge, including on task abort. - let stream = attach::phase("raw terminal stream", attach::GATEWAY, async { + // Cancelling this upgrade does not acknowledge completion of remote owner cleanup; + // unlike capability issuance, its timeout must not trigger another attachment. + let stream = attach::traced_phase(trace, attach::SetupPhase::Upgrade.name(), attach::GATEWAY, async { client.raw_terminal_stream(&attachment).await .map_err(|error| format!("{route}: {}", error.human())) }).await?; diff --git a/crates/fractal-tui/src/ui/live_attach.rs b/crates/fractal-tui/src/ui/live_attach.rs index daaa4d22067..cc9d562f28d 100644 --- a/crates/fractal-tui/src/ui/live_attach.rs +++ b/crates/fractal-tui/src/ui/live_attach.rs @@ -1,6 +1,7 @@ //! Initial terminal attachment lifecycle; callbacks are fenced before any selected-view effects. use super::{Command, Direct, Fetched, Following, Ui, connecting_text, screen_lines}; use crate::feed::TerminalUpdate; +use crate::ui::attach_trace::Trace; use std::{collections::BTreeMap, future::Future, net::Shutdown, sync::{Arc, atomic::{AtomicBool, Ordering}}, time::{Duration, Instant}}; use tokio::{sync::mpsc, task::JoinHandle}; @@ -17,10 +18,64 @@ pub(super) const GATEWAY: Duration = Duration::from_secs(8); pub(super) const DEADLINE: Duration = Duration::from_secs(DISCOVERY.as_secs() + DIRECT.as_secs() + GATEWAY.as_secs() * 2 + 3); pub(super) const RETRY_DELAYS: [Duration; 2] = [Duration::from_secs(1), Duration::from_secs(2)]; -pub(super) async fn phase(name: &str, budget: Duration, future: impl Future>) -> Result { - tokio::time::timeout(budget, future).await.map_err(|_| format!("{name} timed out"))? +/// Only capability issuance is safe to repeat without a remote cancellation receipt: +/// it creates an expiring capability, never an owner PTY attachment. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum SetupPhase { + Capability, + Upgrade, + FirstFrame, } +impl SetupPhase { + pub(super) fn name(self) -> &'static str { + match self { + Self::Capability => "raw terminal capability", + // Authentication and owner-open happen before the server answers the upgrade. + Self::Upgrade => "raw terminal connect/upgrade (including auth and owner-open)", + Self::FirstFrame => "raw terminal first-frame", + } + } + + pub(super) fn timeout(reason: &str) -> Option { + let name = reason.strip_suffix(" timed out")?; + [Self::Capability, Self::Upgrade, Self::FirstFrame] + .into_iter() + .find(|phase| name == phase.name()) + } + + pub(super) fn retry_safe(self) -> bool { + self == Self::Capability + } +} + +#[cfg(test)] +pub(super) async fn phase(name: &'static str, budget: Duration, future: impl Future>) -> Result { + traced_phase(None, name, budget, future).await +} + +pub(super) async fn traced_phase(trace: Option<&Trace>, name: &'static str, budget: Duration, future: impl Future>) -> Result { + let guard = trace.map(|trace| trace.start_phase(name)); + let result = tokio::time::timeout(budget, future).await; + let outcome = match &result { Err(_) => "timeout", Ok(Err(_)) => "failure", Ok(Ok(_)) => "success" }; + if let Some(guard) = guard { guard.end(outcome); } + result.map_err(|_| format!("{name} timed out"))? +} +/// Own cancellation even when a spawned setup future is never polled. +pub(super) struct SetupOwner(Option); + +impl SetupOwner { + pub(super) fn new(trace: Option) -> Self { Self(trace) } + pub(super) fn handed_off(&mut self) { self.0 = None; } +} + +impl Drop for SetupOwner { + fn drop(&mut self) { + if let Some(trace) = &self.0 { trace.finish("setup-dropped"); } + } +} + + #[derive(Default)] pub(super) struct State { generation: u64, @@ -32,16 +87,23 @@ pub(super) struct State { wanted: BTreeMap>, pub(super) tasks: BTreeMap>, pub(super) opens: BTreeMap, + traces: BTreeMap, } impl Drop for State { fn drop(&mut self) { + for trace in self.traces.values() { trace.finish("state-dropped"); } for task in self.tasks.values() { task.abort(); } } } impl State { pub(super) fn cancel(&mut self, agent: &str) { + self.cancel_reason(agent, "cancelled"); + } + + fn cancel_reason(&mut self, agent: &str, reason: &'static str) { + if let Some(trace) = self.traces.remove(agent) { trace.finish(reason); } self.pending.remove(agent); if let Some(task) = self.tasks.get(agent) { task.abort(); } if let Some(wanted) = self.wanted.remove(agent) { wanted.store(false, Ordering::Relaxed); } @@ -51,7 +113,7 @@ impl State { if self.selected.as_deref() == selected { return; } self.generation += 1; if let Some(previous) = self.selected.take() { - self.cancel(&previous); + self.cancel_reason(&previous, "selection-changed"); if let Some(view) = ui.terminal_view_mut(&previous) && view.native.is_none() && view.stale.as_deref().is_some_and(|status| status.starts_with("connecting")) @@ -79,9 +141,12 @@ impl State { self.opens.insert(agent.to_owned(), Instant::now()); let wanted = Arc::new(AtomicBool::new(true)); self.wanted.insert(agent.to_owned(), wanted.clone()); + self.traces.insert(agent.to_owned(), Trace::new(agent)); (self.generation, wanted, serial) } + pub(super) fn trace(&self, agent: &str) -> Option { self.traces.get(agent).cloned() } + pub(super) fn resume_fallback(&mut self, view: &mut crate::ui::TerminalView, runtime_ids: &[String], commands: &mpsc::UnboundedSender) { // A shell has no runtime-resolved fallback. Only Ctrl+R may restart its failed setup. if view.agent.starts_with("terminal/") || view.native.is_some() @@ -118,6 +183,7 @@ impl State { pub(super) fn tick(&mut self, ui: &mut Ui) { for (agent, task) in &self.tasks { if ui.terminal_view(agent).is_none() { + if let Some(trace) = self.traces.remove(agent) { trace.finish("view-closed"); } task.abort(); self.pending.remove(agent); if let Some(wanted) = self.wanted.remove(agent) { wanted.store(false, Ordering::Relaxed); } @@ -127,7 +193,7 @@ impl State { let Some(started) = self.started else { return; }; if started.elapsed() >= DEADLINE { if let Some(agent) = self.selected.take() { - self.cancel(&agent); + self.cancel_reason(&agent, "outer-deadline"); if let Some(view) = ui.terminal_view_mut(&agent) { view.stale = Some(failed("attach setup timed out")); } self.selected = Some(agent); } @@ -161,9 +227,10 @@ impl State { if let Some(view) = ui.terminal_view_mut(&agent).filter(|view| view.native.is_none()) { view.name = format!("{} · {route}", view.name); view.title = view.name.clone(); - view.native = Some(crate::ui::pty::NativeTerminal::spawn(direct.stream, &direct.name, direct.incarnation, rows, columns)); + view.native = Some(crate::ui::pty::NativeTerminal::spawn(direct.stream, &direct.name, direct.incarnation, rows, columns, self.traces.get(&agent).cloned())); view.stale = None; } else { + if let Some(trace) = self.traces.remove(&agent) { trace.finish("view-unavailable"); } close(direct); } if current { ui.flash(format!("Attached · {route}")); } @@ -174,6 +241,7 @@ impl State { } Fetched::NativeFailed { agent, generation, runtime_ids, reason } => { if !self.current(&agent, generation) { return; } + if let Some(trace) = self.traces.remove(&agent) { trace.finish("setup-failed"); } self.note = None; self.started = None; if self.opens.remove(&agent).is_some() { crate::ui::slo::METRICS.add("terminal_open_errors", 1); } @@ -276,6 +344,72 @@ mod tests { use crate::ui::{Pane, TerminalView, glass::Open}; use std::os::unix::net::UnixStream; + #[test] + fn dropping_unpolled_setup_finishes_while_state_retains_trace() { + let (trace, events) = Trace::captured("agent/A"); + let owner = SetupOwner::new(Some(trace.clone())); + let future = async move { let _owner = owner; std::future::pending::<()>().await; }; + drop(future); + trace.finish("state-dropped"); + let events = events.lock().unwrap(); + assert_eq!(events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] != "start").count(), 1); + assert_eq!(events.last().unwrap()["outcome"], "setup-dropped"); + } + + fn poll_setup(future: std::pin::Pin<&mut impl Future>>) { + let waker = std::task::Waker::noop(); + let mut context = std::task::Context::from_waker(waker); + assert!(future.poll(&mut context).is_pending()); + } + + #[tokio::test] + async fn every_setup_phase_expires_with_correlated_outcome() { + for name in ["terminal discovery", "direct open", SetupPhase::Capability.name(), SetupPhase::Upgrade.name(), SetupPhase::FirstFrame.name()] { + let (trace, events) = Trace::captured("agent/test"); + // A zero budget and a permanently pending operation deterministically expires; + // no wall-clock sleep or Tokio test-util feature is needed. + let result = traced_phase(Some(&trace), name, Duration::ZERO, std::future::pending::>()).await; + assert_eq!(result.unwrap_err(), format!("{name} timed out")); + trace.finish("setup-failed"); + let events = events.lock().unwrap(); + assert!(events.iter().any(|event| event["phase"] == name && event["outcome"] == "timeout" && event["level"] == "INFO")); + assert!(events.iter().all(|event| event["attempt_id"] == events[0]["attempt_id"] && event["agent"] == "agent/test")); + } + } + + #[tokio::test] + async fn outer_deadline_ends_every_active_phase_once() { + for name in ["terminal discovery", "direct open", SetupPhase::Capability.name(), SetupPhase::Upgrade.name(), SetupPhase::FirstFrame.name()] { + let (mut state, mut ui, _, _) = selected("agent/A"); + let (trace, events) = Trace::captured("agent/A"); + state.traces.insert("agent/A".into(), trace.clone()); + let mut setup = Box::pin(traced_phase(Some(&trace), name, DEADLINE, std::future::pending::>())); + poll_setup(setup.as_mut()); + state.started = Some(Instant::now() - DEADLINE); + state.tick(&mut ui); + drop(setup); + let events = events.lock().unwrap(); + assert_eq!(events.iter().filter(|event| event["phase"] == name && event["outcome"] != "start").count(), 1); + assert!(events.iter().any(|event| event["phase"] == name && event["outcome"] == "outer-deadline")); + assert_eq!(events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] == "outer-deadline").count(), 1); + } + } + + #[tokio::test] + async fn selecting_another_agent_finishes_before_setup_abort() { + let (mut state, mut ui, _, _) = selected("agent/A"); + let (trace, events) = Trace::captured("agent/A"); + state.traces.insert("agent/A".into(), trace.clone()); + let mut setup = Box::pin(traced_phase(Some(&trace), SetupPhase::Upgrade.name(), GATEWAY, std::future::pending::>())); + poll_setup(setup.as_mut()); + state.select(&mut ui, Some("agent/B")); + drop(setup); + trace.finish("dropped"); + let events = events.lock().unwrap(); + assert_eq!(events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] != "start").count(), 1); + assert!(events.iter().any(|event| event["phase"] == SetupPhase::Upgrade.name() && event["outcome"] == "selection-changed")); + } + fn view(agent: &str) -> TerminalView { TerminalView { agent: agent.into(), title: agent.into(), name: agent.into(), lines: Vec::new(), cursor: None, stale: Some("connecting".into()), ended: None, native: None } } @@ -367,9 +501,9 @@ mod tests { let generation = state.generation; state.apply_fetched(&mut ui, Fetched::NativeFailed { agent: "agent/A".into(), generation, runtime_ids: vec!["runtime/A".into()], - reason: "raw terminal attachment timed out".into(), + reason: format!("{} timed out", SetupPhase::Upgrade.name()), }, &tx); - assert_eq!(ui.terminal.as_ref().unwrap().stale.as_deref(), Some("failed: raw terminal attachment timed out · Ctrl+R to retry")); + assert_eq!(ui.terminal.as_ref().unwrap().stale.as_deref(), Some("failed: raw terminal connect/upgrade (including auth and owner-open) timed out · Ctrl+R to retry")); assert!(state.started.is_none()); assert!(state.note.is_none()); assert!(rx.try_recv().is_err()); diff --git a/crates/fractal-tui/src/ui/mod.rs b/crates/fractal-tui/src/ui/mod.rs index 7b468cde02e..293fce7c850 100644 --- a/crates/fractal-tui/src/ui/mod.rs +++ b/crates/fractal-tui/src/ui/mod.rs @@ -40,6 +40,7 @@ pub use lastrun::terminal_gone as lastrun_terminal_gone; mod glass_store; mod lastrun; pub(crate) mod open_timing; +pub(crate) mod attach_trace; pub mod layout; pub mod live; pub(crate) mod slo; diff --git a/crates/fractal-tui/src/ui/pty.rs b/crates/fractal-tui/src/ui/pty.rs index d2d50388a7a..b9f45f56ee7 100644 --- a/crates/fractal-tui/src/ui/pty.rs +++ b/crates/fractal-tui/src/ui/pty.rs @@ -3,7 +3,7 @@ //! emulator here. The PTY takes the pane's size, keys go to it as bytes, and the history that //! scrolls off the top stays here to scroll back through. -use super::theme; +use super::{attach_trace::Trace, theme}; #[path = "pty_graphics.rs"] mod graphics; #[path = "pty_connection.rs"] @@ -186,6 +186,9 @@ pub(crate) struct NativeTerminal { presented: AtomicBool, cell: Mutex, painter: Mutex, + trace: Option, + /// A decoded screen was drawn in this frame, awaiting a successful outer flush. + pending_render: AtomicBool, } impl NativeTerminal { @@ -198,13 +201,14 @@ impl NativeTerminal { incarnation: String, rows: u16, columns: u16, + trace: Option, ) -> Self { let size = Size { rows: rows.max(1), columns: columns.max(1), }; let screen = Arc::new(Mutex::new(Screen::new(size, Requests::default()))); - let input = start(stream, name, size, &screen); + let input = start(stream, name, size, &screen, trace.clone()); Self { screen, input: Mutex::new(input), @@ -213,6 +217,8 @@ impl NativeTerminal { presented: AtomicBool::new(false), cell: Mutex::new(pty_terminal::CellSize::default()), painter: Mutex::new(graphics::Painter::default()), + trace, + pending_render: AtomicBool::new(false), } } @@ -227,7 +233,7 @@ impl NativeTerminal { screen.retry_at = None; screen.attached = false; } - let input = start(stream, name, size, &self.screen); + let input = start(stream, name, size, &self.screen, None); *self.input.lock().unwrap_or_else(|error| error.into_inner()) = input; } @@ -531,6 +537,20 @@ impl NativeTerminal { self.presented.load(Ordering::Relaxed) } + /// Clear the draw acknowledgment before each outer frame, including hidden panes. + pub(crate) fn frame_started(&self) { + self.pending_render.store(false, Ordering::Relaxed); + } + + /// Acknowledge only a screen drawn during the successfully flushed outer frame. + pub(crate) fn frame_rendered(&self) { + if self.pending_render.swap(false, Ordering::Relaxed) + && let Some(trace) = &self.trace + { + trace.complete(); + } + } + pub(crate) fn attached(&self) -> bool { self.lock().attached } @@ -597,6 +617,14 @@ impl NativeTerminal { } buf.set_line(area.x, area.y + row as u16, &Line::from(spans), area.width); } + if screen.attached + && screen.ended.is_none() + && screen.lost.is_none() + && rows > 0 + && columns > 0 + { + self.pending_render.store(true, Ordering::Relaxed); + } // The cursor, when the bottom is shown and the program shows it: the person's own // terminal cursor where it has the keys, so it keeps the shape the program asked for // (vim's bar while inserting), blinks, and places an input method's text; otherwise @@ -627,6 +655,14 @@ impl NativeTerminal { } } +impl Drop for NativeTerminal { + fn drop(&mut self) { + if let Some(trace) = &self.trace { + trace.finish("native-dropped"); + } + } +} + /// The cursor style until a program asks for one: the person's own shape is kept then. const UNASKED: CursorStyle = CursorStyle { shape: CursorShape::HollowBlock, @@ -666,13 +702,14 @@ fn start( name: &str, size: Size, screen: &Arc>, + trace: Option, ) -> mpsc::Sender { let (input, inputs) = mpsc::channel(); let shared = screen.clone(); let name = name.to_owned(); std::thread::Builder::new() .name(format!("pty {name}")) - .spawn(move || run(stream, size, &shared, &inputs)) + .spawn(move || run(stream, size, &shared, &inputs, trace)) .ok(); input } @@ -709,23 +746,42 @@ fn run( size: Size, screen: &Mutex, inputs: &mpsc::Receiver, + trace: Option, ) { let lock = || screen.lock().unwrap_or_else(|error| error.into_inner()); if let Err(error) = stream.set_nonblocking(false) { lock().ended = Some(error.to_string()); + if let Some(trace) = &trace { + trace.finish("native-attach-failed"); + } return; } let mut control = stream.try_clone(); let mut requested = size; + let screen_phase = trace.as_ref().map(|trace| trace.start_phase("native-first-screen")); let mut connection = match connection::Connection::attach_over( stream, size.rows, size.columns, ATTACH_TIMEOUT, + trace.clone(), ) { - Ok(connection) => connection, + Ok(connection) => { + if let Some(phase) = screen_phase { phase.end("success"); } + connection + } Err(error) => { + if let Some(phase) = screen_phase { + phase.end(if error.kind() == std::io::ErrorKind::TimedOut { "timeout" } else { "failure" }); + } dropped(&mut lock(), error.to_string()); + if let Some(trace) = &trace { + trace.finish(if error.kind() == std::io::ErrorKind::TimedOut { + "native-attach-timeout" + } else { + "native-attach-failed" + }); + } return; } }; @@ -830,15 +886,24 @@ fn run( Ok(Some(SessionEvent::Exit(code))) => { super::slo::METRICS.add("dead_terminal_tabs", 1); lock().ended = Some(format!("the program exited ({code})")); + if let Some(trace) = &trace { + trace.finish("native-exited"); + } return; } Ok(Some(SessionEvent::Closed)) => { dropped(&mut lock(), "the connection closed".into()); + if let Some(trace) = &trace { + trace.finish("native-stream-lost"); + } return; } Ok(None) => {} Err(error) => { dropped(&mut lock(), error.to_string()); + if let Some(trace) = &trace { + trace.finish("native-stream-lost"); + } return; } } @@ -1178,6 +1243,102 @@ mod tests { KeyEvent::new(code, modifiers) } + #[test] + fn screen_milestones_require_decoding_drawing_and_successful_flush_per_attempt() { + use pty_core::protocol::{MessageType, encode_packet}; + use std::io::Write as _; + + let (first_trace, first_events) = Trace::captured("agent/first"); + let (second_trace, second_events) = Trace::captured("agent/second"); + let (first_socket, mut first_peer) = UnixStream::pair().unwrap(); + let (second_socket, mut second_peer) = UnixStream::pair().unwrap(); + let first = NativeTerminal::spawn(first_socket, "first", "one".into(), 4, 20, Some(first_trace)); + let second = NativeTerminal::spawn(second_socket, "second", "two".into(), 4, 20, Some(second_trace)); + let area = Rect::new(0, 0, 20, 4); + let mut buffer = Buffer::empty(area); + let rendered = |events: &Arc>>| { + events.lock().unwrap().iter().filter(|event| event["phase"] == "first-screen-rendered").count() + }; + + // Drawing the pending emulator and acknowledging an empty frame are not a screen. + first.frame_started(); + first.draw(&mut buffer, area, false); + first.frame_rendered(); + assert_eq!(rendered(&first_events), 0); + first_peer.write_all(&pty_core::protocol::encode_geometry(4, 20)).unwrap(); + let geometry_deadline = Instant::now() + Duration::from_secs(5); + while !first_events.lock().unwrap().iter().any(|event| event["phase"] == "first-frame-decoded") { + assert!(Instant::now() < geometry_deadline, "local geometry was not decoded"); + std::thread::sleep(Duration::from_millis(1)); + } + first.frame_started(); + first.draw(&mut buffer, area, false); + first.frame_rendered(); + assert_eq!(rendered(&first_events), 0, "geometry alone is not a screen"); + first_peer.write_all(&encode_packet(MessageType::Screen, b"first screen")).unwrap(); + second_peer.write_all(&encode_packet(MessageType::Screen, b"second screen")).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while !first.attached() || !second.attached() { + assert!(Instant::now() < deadline, "local snapshot was not decoded"); + std::thread::sleep(Duration::from_millis(1)); + } + assert_eq!(rendered(&first_events), 0); + assert_eq!(rendered(&second_events), 0); + first.frame_started(); + first.draw(&mut buffer, Rect::new(0, 0, 0, 4), false); + first.frame_rendered(); + assert_eq!(rendered(&first_events), 0); + + // A failed outer flush has no acknowledgment. A later hidden frame must not + // accidentally acknowledge the stale buffer draw from that failed frame. + first.frame_started(); + first.draw(&mut buffer, area, false); + first.frame_started(); + first.frame_rendered(); + assert_eq!(rendered(&first_events), 0); + + second.frame_started(); + second.draw(&mut buffer, area, false); + assert_eq!(rendered(&second_events), 0, "buffer draw is not an outer flush"); + second.frame_rendered(); + assert_eq!(rendered(&second_events), 1); + assert_eq!(rendered(&first_events), 0, "another attempt must not complete this one"); + first.frame_started(); + first.draw(&mut buffer, area, true); + first.frame_rendered(); + first.frame_rendered(); + assert_eq!(rendered(&first_events), 1); + drop(first); + drop(second); + + let first_events = first_events.lock().unwrap(); + let second_events = second_events.lock().unwrap(); + assert_ne!(first_events[0]["attempt_id"], second_events[0]["attempt_id"]); + for (events, agent) in [(&*first_events, "agent/first"), (&*second_events, "agent/second")] { + assert!(events.iter().all(|event| event["agent"] == agent && event["attempt_id"] == events[0]["attempt_id"])); + let decoded = events.iter().position(|event| event["phase"] == "first-frame-decoded").unwrap(); + let rendered = events.iter().position(|event| event["phase"] == "first-screen-rendered").unwrap(); + assert!(decoded < rendered); + assert_eq!(events.iter().filter(|event| event["phase"] == "first-frame-decoded").count(), 1); + let terminal: Vec<_> = events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] != "start").collect(); + assert_eq!(terminal.len(), 1); + assert_eq!(terminal[0]["outcome"], "success"); + } + } + + #[test] + fn dropping_native_during_setup_finishes_without_waiting_for_worker() { + let (trace, events) = Trace::captured("agent/dropped"); + let (socket, _peer) = UnixStream::pair().unwrap(); + let terminal = NativeTerminal::spawn(socket, "test", "one".into(), 4, 20, Some(trace)); + drop(terminal); + let events = events.lock().unwrap(); + let terminal: Vec<_> = events.iter().filter(|event| event["phase"] == "attempt" && event["outcome"] != "start").collect(); + assert_eq!(terminal.len(), 1); + assert_eq!(terminal[0]["outcome"], "native-dropped"); + assert!(!events.iter().any(|event| event["phase"] == "first-screen-rendered")); + } + #[test] fn tries_to_attach_again_wait_longer_each_time_and_never_more_than_the_most() { let waits: Vec = (0..8).map(|tries| retry_delay(tries).as_secs()).collect(); @@ -1223,7 +1384,7 @@ mod tests { let (tui, mut daemon) = UnixStream::pair().unwrap(); daemon.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); - let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30); + let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30, None); let mut reader = PacketReader::new(); let mut pending = std::collections::VecDeque::new(); let mut next = |daemon: &mut UnixStream| loop { @@ -1307,7 +1468,7 @@ mod tests { .collect::() }; let (tui, mut daemon) = UnixStream::pair().unwrap(); - let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 6, 30); + let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 6, 30, None); let mut bytes = [0_u8; 64]; let _ = daemon.read(&mut bytes).unwrap(); daemon @@ -1380,7 +1541,7 @@ mod tests { buf }; let (tui, mut daemon) = UnixStream::pair().unwrap(); - let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 4, 10); + let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 4, 10, None); let mut bytes = [0_u8; 64]; let _ = daemon.read(&mut bytes).unwrap(); daemon @@ -1453,7 +1614,7 @@ mod tests { use pty_core::protocol::{MessageType, PacketReader, encode_packet}; use std::io::{Read as _, Write as _}; let (tui, mut daemon) = UnixStream::pair().unwrap(); - let terminal = NativeTerminal::spawn(tui, "test", "incarnation".into(), 4, 20); + let terminal = NativeTerminal::spawn(tui, "test", "incarnation".into(), 4, 20, None); let mut reader = PacketReader::new(); let mut packets = Vec::new(); let mut next = |daemon: &mut UnixStream, packets: &mut Vec<_>| { diff --git a/crates/fractal-tui/src/ui/pty_connection.rs b/crates/fractal-tui/src/ui/pty_connection.rs index c6baf7f31fa..0bebdb099b6 100644 --- a/crates/fractal-tui/src/ui/pty_connection.rs +++ b/crates/fractal-tui/src/ui/pty_connection.rs @@ -1,5 +1,6 @@ //! Viewer-side framing keeps the initial snapshot's geometry at its wire boundary. //! SessionConnection's effective dimensions include read-ahead frames after SCREEN. +use super::Trace; use pty_client::connection::SessionEvent; use pty_core::protocol::{self, MessageType, PacketReader}; use std::collections::VecDeque; @@ -14,6 +15,7 @@ pub(super) struct Connection { screen: Vec, rows: u16, cols: u16, + trace: Option, } impl Connection { @@ -22,6 +24,7 @@ impl Connection { rows: u16, cols: u16, timeout: Duration, + trace: Option, ) -> io::Result { socket.write_all(&protocol::encode_attach(rows, cols))?; let mut connection = Self { @@ -31,6 +34,7 @@ impl Connection { screen: Vec::new(), rows, cols, + trace, }; let deadline = Instant::now() + timeout; let mut before_screen = VecDeque::new(); @@ -91,6 +95,9 @@ impl Connection { MessageType::Exit => SessionEvent::Exit(protocol::decode_exit(&packet.payload)), _ => continue, }; + if let Some(trace) = self.trace.take() { + trace.milestone("first-frame-decoded"); + } self.pending.push_back(event); } Ok(self.pending.pop_front()) @@ -117,3 +124,41 @@ impl Connection { impl Drop for Connection { fn drop(&mut self) { self.disconnect(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bytes_and_partial_frames_do_not_claim_a_decoded_frame() { + let (trace, events) = Trace::captured("agent/fragmented"); + let (socket, mut peer) = UnixStream::pair().unwrap(); + let mut connection = Connection { + socket, + reader: PacketReader::new(), + pending: VecDeque::new(), + screen: Vec::new(), + rows: 4, + cols: 20, + trace: Some(trace.clone()), + }; + let geometry = protocol::encode_geometry(4, 20); + let split = geometry.len() - 1; + peer.write_all(&geometry[..split]).unwrap(); + assert!(connection.next_event(Some(Duration::from_secs(1))).unwrap().is_none()); + assert_eq!(events.lock().unwrap().len(), 1, "raw bytes are not decoded frames"); + peer.write_all(&geometry[split..]).unwrap(); + assert!(matches!(connection.next_event(Some(Duration::from_secs(1))).unwrap(), + Some(SessionEvent::Geometry { rows: 4, cols: 20 }))); + { + let events = events.lock().unwrap(); + assert_eq!(events.len(), 2); + assert_eq!(events[1]["phase"], "first-frame-decoded"); + assert!(!events.iter().any(|event| event["phase"] == "first-screen-rendered")); + } + peer.write_all(&protocol::encode_packet(MessageType::Screen, b"snapshot")).unwrap(); + assert!(matches!(connection.next_event(Some(Duration::from_secs(1))).unwrap(), + Some(SessionEvent::Screen(_)))); + assert_eq!(events.lock().unwrap().len(), 2, "decoding later frames is one-shot"); + } +} diff --git a/crates/fractal-tui/src/ui/pty_geometry_tests.rs b/crates/fractal-tui/src/ui/pty_geometry_tests.rs index 653be09125c..bce5e2a0b24 100644 --- a/crates/fractal-tui/src/ui/pty_geometry_tests.rs +++ b/crates/fractal-tui/src/ui/pty_geometry_tests.rs @@ -54,7 +54,7 @@ fn coalesced_initial_snapshot_uses_its_geometry_before_a_later_shrink() { // Queue the entire stream before the attach thread can read it: the first // socket read sees both geometries, not just the one belonging to SCREEN. assert_eq!(daemon.write(&coalesced).unwrap(), coalesced.len()); - let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30); + let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30, None); let mut reader = PacketReader::new(); let mut pending = VecDeque::new(); assert_eq!( @@ -104,7 +104,7 @@ fn coalesced_initial_snapshot_uses_its_geometry_before_a_later_shrink() { fn cell_metrics_preserve_requested_dimensions_under_shared_geometry_constraints() { let (tui, mut daemon) = UnixStream::pair().unwrap(); daemon.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); - let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30); + let terminal = NativeTerminal::spawn(tui, "test", "one".into(), 10, 30, None); let mut reader = PacketReader::new(); let mut pending = VecDeque::new(); let attach = next_packet(&mut daemon, &mut reader, &mut pending); diff --git a/crates/fractal-tui/src/ui/recent.rs b/crates/fractal-tui/src/ui/recent.rs index b7fd78c9d62..02fa334c35d 100644 --- a/crates/fractal-tui/src/ui/recent.rs +++ b/crates/fractal-tui/src/ui/recent.rs @@ -867,7 +867,7 @@ mod tests { let (client, mut peer) = UnixStream::pair().unwrap(); peer.set_read_timeout(Some(Duration::from_secs(3))).unwrap(); ui.terminal_view_mut(&id).unwrap().native = - Some(NativeTerminal::spawn(client, &id, "one".into(), 24, 80)); + Some(NativeTerminal::spawn(client, &id, "one".into(), 24, 80, None)); peer.write_all(&encode_packet( MessageType::Screen, b"\x1b[?1004h\x1b[?1003h\x1b[?1006h\x1b[?2004h", diff --git a/crates/fractal-tui/tests/support/attach_default.rs b/crates/fractal-tui/tests/support/attach_default.rs index 705270e479d..785154315fd 100644 --- a/crates/fractal-tui/tests/support/attach_default.rs +++ b/crates/fractal-tui/tests/support/attach_default.rs @@ -116,6 +116,7 @@ fn attach_default_probe_worker() -> Result<()> { "one".into(), 24, 80, + None, )), }); } @@ -211,7 +212,7 @@ fn attach_redraw_cost() { cursor: None, stale: None, ended: None, - native: Some(pty::NativeTerminal::spawn(tui, "bench", "one".into(), 38, 120)), + native: Some(pty::NativeTerminal::spawn(tui, "bench", "one".into(), 38, 120, None)), }); if std::env::var_os("FRACTAL_TUI_BENCH_ALERT").is_some() { ui.world.attention = Load::Ready(vec![ask(AGENT)]); diff --git a/crates/fractal-tui/tests/support/terminal_tab.rs b/crates/fractal-tui/tests/support/terminal_tab.rs index 0391476eac2..a11b4eab530 100644 --- a/crates/fractal-tui/tests/support/terminal_tab.rs +++ b/crates/fractal-tui/tests/support/terminal_tab.rs @@ -87,6 +87,7 @@ fn terminal_tab_probe_worker() -> Result<()> { "one".into(), 24, 80, + None, )), }); let mut last = String::new(); diff --git a/docs/when-something-is-wrong.md b/docs/when-something-is-wrong.md index 678ae806431..06612c3edca 100644 --- a/docs/when-something-is-wrong.md +++ b/docs/when-something-is-wrong.md @@ -39,6 +39,95 @@ fractal terminals peek agent/garden/worker The last terminal screen can explain a login, crash, or provider limit; the graph says whether the stop was requested, the owner ended, a suspension is active, or the runtime failed. A stopped agent can retain a historical terminal, and an agent with no recorded terminal may have nothing to peek. +## Raw terminal setup stalls + +Raw terminal setup logs bounded DEBUG phase events, not terminal contents or +capabilities. The API handler records its start timestamp (`started_at_unix_ms`) +and monotonic `elapsed_ms` for `auth`, `lookup`, `capability`, `owner-open`, and +`upgrade`; completed lookup, capability, and open phases also record +`duration_ms`. Here `auth` is the handler's scope/person check, after transport +authentication. For remote ATTACH, the owner peer separately records +`peer-auth`, `owner-attachment`, `owner-open`, and `peer-upgrade`. PEEK uses its +existing owner lease-admission path instead. A phase start without completion +locates the unfinished or rejected operation; it does not prove a timeout cause. +After upgrade, `first-frame` records the first successfully sent PTY byte chunk, +with elapsed time measured from the splice's start and the terminal/incarnation +in its `raw_terminal_stream` span. There are no per-frame debug events. + +The TUI records an attempt UUID and target agent on every phase event. Its +attempt-scoped lifecycle records a terminal outcome with the phase reached and +elapsed time even on outer deadline expiry, selection changes, or drop; failures +are INFO-level outcomes rather than completion-only DEBUG messages. Capability +issuance and combined client connect/upgrade retain their 8-second budgets. +Only a capability-issuance timeout retries automatically, at most twice after +1 and 2 seconds, keeping the selected incarnation. Each cancelled issuance +future is dropped before retry; an unused capability opens no PTY and expires +after 60 seconds. Upgrade timeouts name that phase and offer Ctrl-R, without an +automatic retry. The overall setup deadline and single-flight guard still apply. + +Without any opt-in variable, every non-successful terminal attempt also emits +one `tracing::info!` event and is retained in +`$XDG_CACHE_HOME/st3/stui/attach.log` (otherwise +`~/.cache/st3/stui/attach.log`), beside `slow.log` and `exits.log`. This uses the +same cache-log permissions and 100 KB reset bound as those existing logs. +It records only attempt ID, target agent, phase reached, outcome, and timing; +it requires no tracing subscriber or OTEL exporter. Start and success events +remain exclusive to the opt-in sink described below. + +For a private, device-local reproduction log, launch the TUI with an absolute +JSONL destination: + +```sh +FRACTAL_TUI_ATTACH_TRACE_FILE="$HOME/raw-attach.jsonl" fractal ui --client +``` + +This file is the actual client diagnostic sink (`service="fractal-tui"`), not an +OTEL export or daemon journal. It contains identifiers and timing, not terminal +contents, credentials, or capabilities; treat it as private. Open the target pane +without typing provider input. `first-frame-decoded` means the client decoded a +PTY protocol frame; `first-screen-rendered` follows a native screen draw and +successful synchronized terminal-frame completion. Neither is inferred from +server first-byte timing. + +Use a separate file per TUI process. New files have mode 0600; an existing file +must be a regular, single-link, non-symlink file owned by the current user with +mode 0600, or logging stays disabled. Appends stop at 4 MiB without truncation. +The sink is best-effort on I/O errors, size exhaustion, or lock contention; it +never waits for another process's log lock. Records include `at_unix_ms` for +server correlation and monotonic elapsed time for local phase duration. + +Server phase DEBUG events use the daemon's stderr sink. Under a user service, +read its journal (for the standard deployment, +`journalctl --user -u smalltalk.service`). Enable the relevant modules in the +managed daemon's `RUST_LOG`, for example +`info,fractal_api_client_v0::api::client_v0::raw_terminal=debug,fractal_api_core::api::client_v0::raw_terminal=debug,fractal_peer::peer=debug`. +Changing a running service's environment requires its normal authorized +deployment; the TUI does not change daemon logging or restart it. Existing OTEL +request spans use `service.name="st-daemon"`; DEBUG stderr events are not exported +by its INFO-and-above OTEL log filter. Correlate server terminal/incarnation and +request times separately from the client attempt UUID. + +Endpoint resolution, Unix connect, and WebSocket upgrade are private generated +client boundaries: they are **not separable without codegen** from this caller. +For remote connections the generated connector also combines DNS/TCP/TLS and +upgrade. Separate client-side connect/upgrade and first-binary-arrival timing +must be added at the next client regeneration; server `first-frame` is a send +measurement, not proof that the TUI received or rendered a screen. + +The HTTP upgrade includes server authority checks, runtime lookup, and opening +the owner's transport. A client-side upgrade timeout cannot identify which of +these operations stalled. Runtime lookup and local PTY opening retain their +incarnation fences: the local socket must prove the selected daemon PID and +registry creation time before any PTY frame is sent. + +Dropping an in-progress HTTP upgrade closes the client's connection; it is not a +remote cancellation acknowledgment. The remote request can still be opening its +owner transport. Failed upgrades drop their captured transports, and established +bridges propagate connector closure, but there is no acknowledgment that all +remote cleanup completed before another attempt. Do not automatically retry an +upgrade timeout on the assumption that dropping its future completed remote +cleanup. Keep provider input out of setup diagnostics. + ## Linux PID namespaces and procfs The runtime requires `/proc` to describe its own PID namespace. Entering a child