Label raw terminal setup phases and retry capability issuance

crates/fractal-api-client-v0/src/api/client_v0/raw_terminal.rs

@@ -134,8 +134,18 @@ pub async fn stream(
     Query(query): Query<StreamQuery>,
     headers: HeaderMap,
 ) -> Result<Response, ApiError> {
+    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)]

crates/fractal-api-core/src/api/client_v0/raw_terminal.rs

@@ -34,6 +34,8 @@ pub async fn splice(
     _upgrade: Option<opentelemetry::trace::SpanContext>,
     mut first_frame: Option<tracing::Span>,
 ) {
+    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;
                 }
             }

crates/fractal-peer/src/peer.rs

@@ -1190,6 +1190,8 @@ async fn receive_raw_terminal(
     axum::Extension(splice): axum::Extension<RawSplice>,
     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
 }

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"

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<Inner>);
+
+struct Inner {
+    attempt: uuid::Uuid,
+    agent: String,
+    started: Instant,
+    state: Mutex<State>,
+}
+
+#[derive(Default)]
+struct State {
+    finished: bool,
+    next: u64,
+    last_phase: Option<&'static str>,
+    active: BTreeMap<u64, (&'static str, Instant)>,
+    milestones: BTreeSet<&'static str>,
+    sink: Option<File>,
+    #[cfg(test)]
+    captured: Option<Arc<Mutex<Vec<serde_json::Value>>>>,
+    #[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<Mutex<Vec<serde_json::Value>>>) {
+        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<u128>) {
+        #[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);
+    }
+}

crates/fractal-tui/src/ui/context.rs

@@ -860,6 +860,7 @@ mod tests {
                 "one".into(),
                 24,
                 80,
+                None,
             )),
         });
         peer.write_all(&encode_packet(