From d61f5d15e737c9f42a40e520f170b8d29bf7a1d1 Mon Sep 17 00:00:00 2001 From: Geoffrey Churchill Date: Fri, 23 Oct 2026 14:08:27 -0400 Subject: [PATCH] feat: add `window_title` to `shpool list --json` See discussion #440 --- libshpool/src/daemon/server.rs | 110 ++++++++++++++++------- libshpool/src/daemon/shell.rs | 67 +++++++++++++- libshpool/src/session_restore.rs | 22 +++++ shpool-protocol/src/lib.rs | 6 ++ shpool/tests/list.rs | 148 ++++++++++++++++++++++++++++++- 5 files changed, 316 insertions(+), 37 deletions(-) diff --git a/libshpool/src/daemon/server.rs b/libshpool/src/daemon/server.rs index b12456be..9e83e283 100644 --- a/libshpool/src/daemon/server.rs +++ b/libshpool/src/daemon/server.rs @@ -839,41 +839,80 @@ impl Server { #[instrument(skip_all)] fn handle_list(&self, mut stream: UnixStream) -> anyhow::Result<()> { - let _s = span!(Level::INFO, "lock(shells)").entered(); - let shells = self.shells.lock(); - let sessions: anyhow::Result> = shells - .iter() - .map(|(session_name, session)| { - let status = match session.inner.try_lock() { - Some(_) => SessionStatus::Disconnected, - None => SessionStatus::Attached, - }; + // None if the session's shell->client thread has exited. + type Reply = Option>; - let lifecycle_state = session.lifecycle.snapshot(); - let last_connected_at_unix_ms = lifecycle_state - .last_connected_at - .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64)) - .transpose()?; - let last_disconnected_at_unix_ms = lifecycle_state - .last_disconnected_at - .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64)) - .transpose()?; - - Ok(Session { - name: session_name.to_string(), - started_at_unix_ms: session - .started_at - .duration_since(time::UNIX_EPOCH)? - .as_millis() as i64, - last_connected_at_unix_ms, - last_disconnected_at_unix_ms, - status, - shell_pid: session.child_pid, - attachments: lifecycle_state.attachment.into_iter().collect(), + // Ask every session for its metadata up front, so the shell->client threads + // work on their answers in parallel. Asking never blocks, so it is safe under + // the shells lock. + let sessions: anyhow::Result> = { + let _s = span!(Level::INFO, "lock(shells)").entered(); + let shells = self.shells.lock(); + shells + .iter() + .map(|(session_name, session)| { + let status = match session.inner.try_lock() { + Some(_) => SessionStatus::Disconnected, + None => SessionStatus::Attached, + }; + + let lifecycle_state = session.lifecycle.snapshot(); + let last_connected_at_unix_ms = lifecycle_state + .last_connected_at + .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64)) + .transpose()?; + let last_disconnected_at_unix_ms = lifecycle_state + .last_disconnected_at + .map(|t| t.duration_since(time::UNIX_EPOCH).map(|d| d.as_millis() as i64)) + .transpose()?; + + let (reply_tx, reply_rx) = crossbeam_channel::bounded(1); + let reply = session.fetch_metadata.send(reply_tx).ok().map(|()| reply_rx); + + Ok(( + Session { + name: session_name.to_string(), + started_at_unix_ms: session + .started_at + .duration_since(time::UNIX_EPOCH)? + .as_millis() as i64, + last_connected_at_unix_ms, + last_disconnected_at_unix_ms, + status, + shell_pid: session.child_pid, + attachments: lifecycle_state.attachment.into_iter().collect(), + window_title: None, + }, + reply, + )) }) - }) - .collect(); - let sessions = sessions.context("collecting running session metadata")?; + .collect() + }; + let (mut sessions, replies): (Vec, Vec) = + sessions.context("collecting running session metadata")?.into_iter().unzip(); + + // The lock is released by now, so we can wait. A busy shell->client thread + // (e.g. one building a large restore buffer for a reattach) may miss the + // deadline, in which case its session is listed without its metadata. + let deadline = Instant::now() + SESSION_MSG_TIMEOUT; + for (session, reply) in sessions.iter_mut().zip(replies) { + match reply.map(|reply| reply.recv_deadline(deadline)) { + Some(Ok(shell::SessionMetadata { window_title })) => { + session.window_title = window_title + } + Some(Err(crossbeam_channel::RecvTimeoutError::Timeout)) => { + info!( + "listing {} without metadata, which it did not report in time", + session.name + ) + } + None | Some(Err(crossbeam_channel::RecvTimeoutError::Disconnected)) => info!( + "listing {} without metadata, since its shell->client thread has exited", + session.name + ), + } + } + write_reply(&mut stream, ListReply { sessions })?; Ok(()) } @@ -1196,6 +1235,9 @@ impl Server { // no ack chan so we can afford to buffer a bit. let (maybe_switch_tx, maybe_switch_rx) = crossbeam_channel::bounded(10); + // Unbounded so that sending a metadata request never blocks. + let (fetch_metadata_tx, fetch_metadata_rx) = crossbeam_channel::unbounded(); + let shell_to_client_ctl = Arc::new(Mutex::new(shell::ShellToClientCtl { next_heartbeat_id: std::sync::atomic::AtomicU64::new(0), client_connection: client_connection_tx, @@ -1240,6 +1282,7 @@ impl Server { heartbeat_ack: heartbeat_ack_tx, maybe_switch: maybe_switch_rx, child_exit_notifier: shell_to_client_child_exit_notifier, + fetch_metadata: fetch_metadata_rx, })?); if let Some(ttl_secs) = header.ttl_secs { @@ -1256,6 +1299,7 @@ impl Server { child_exit_notifier, started_at: time::SystemTime::now(), lifecycle: shell::SessionLifecycle::default(), + fetch_metadata: fetch_metadata_tx, inner: Arc::new(Mutex::new(session_inner)), }) } diff --git a/libshpool/src/daemon/shell.rs b/libshpool/src/daemon/shell.rs index bcc7657c..38c5ff6b 100644 --- a/libshpool/src/daemon/shell.rs +++ b/libshpool/src/daemon/shell.rs @@ -15,7 +15,7 @@ use std::{ io, io::{Read, Write}, - net, + iter, net, ops::Add, os::unix::net::UnixStream, sync::{ @@ -114,6 +114,10 @@ pub struct Session { pub child_pid: libc::pid_t, pub child_exit_notifier: Arc, pub shell_to_client_ctl: Arc>, + /// Asks the shell->client thread for the session's [`SessionMetadata`]. It + /// is kept here rather than in [`ShellToClientCtl`] so that requesters + /// don't wait for that lock. + pub fetch_metadata: crossbeam_channel::Sender, pub pager_ctl: Arc>>, /// Mutable state with the lock held by the servicing handle_attach thread /// while a tty is attached to the session. Probing the mutex can be used @@ -245,6 +249,46 @@ pub enum ClientConnectionMsg { Disconnect, } +/// What the shell->client thread returns in response to a +/// [`Session::fetch_metadata`] request. +#[derive(Debug, Clone)] +pub struct SessionMetadata { + /// The window title. Any non-UTF-8 bytes are replaced with U+FFFD. + pub window_title: Option, +} + +/// A request on [`Session::fetch_metadata`]. It is its own reply channel, so +/// each requester gets a fresh private channel. +pub type MetadataRequest = crossbeam_channel::Sender; + +/// Handle `request`, received from `requests`, the shell->client thread's +/// [`Session::fetch_metadata`] channel. Replies to it and to every request +/// already queued behind it, so concurrent requests get answered as a single +/// batch. If the channel has closed because the [`Session`] is gone, `requests` +/// is replaced with [`crossbeam_channel::never`]. +fn handle_fetch_metadata( + spool: &dyn session_restore::SessionSpool, + request: Result, + requests: &mut crossbeam_channel::Receiver, +) { + let Ok(request) = request else { + // A closed channel is always ready, so leaving it in place would make the + // select that produced `request` pick it on every pass. The thread runs on + // until the shell exits or its control channels close. + *requests = crossbeam_channel::never(); + return; + }; + let metadata = SessionMetadata { + window_title: spool + .window_title() + .map(|window_title| String::from_utf8_lossy(window_title).into_owned()), + }; + for reply in iter::once(request).chain(requests.try_iter()) { + // Nobody needs the reply if the requester has stopped waiting. + let _ = reply.send(metadata.clone()); + } +} + pub struct ShellToClientArgs { pub conn_id: usize, pub tty_size: TtySize, @@ -260,6 +304,9 @@ pub struct ShellToClientArgs { // true if the client is still live, false if it has hung up on us. pub heartbeat_ack: crossbeam_channel::Sender<(u64, bool)>, pub child_exit_notifier: Arc, + // The only sending end is the Session's fetch_metadata field, so this closes when the Session + // is dropped. + pub fetch_metadata: crossbeam_channel::Receiver, } impl SessionInner { @@ -291,6 +338,7 @@ impl SessionInner { let mut output_spool = session_restore::new(config, &args.tty_size, args.scrollback_lines, &name); + let mut fetch_metadata = args.fetch_metadata; let mut buf: Vec = vec![0; consts::BUF_SIZE]; let mut poll_fds = [poll::PollFd::new( watchable_master.borrow_fd(), @@ -300,8 +348,18 @@ impl SessionInner { // block until we get the first connection attached so that we don't // drop the initial prompt on the floor info!("waiting for initial client connection"); - let mut client_conn: ClientConnectionMsg = - args.client_connection.recv().context("waiting for initial client connection")?; + let mut client_conn: ClientConnectionMsg = loop { + crossbeam_channel::select! { + recv(args.client_connection) -> client_conn => { + break client_conn.context("waiting for initial client connection")?; + } + // The session is already in `shells`, so metadata requests can arrive before + // its first client. + recv(fetch_metadata) -> request => { + handle_fetch_metadata(&*output_spool, request, &mut fetch_metadata); + } + } + }; args.client_connection_ack .send(ClientConnectionStatus::New) .context("sending initial client connection ack")?; @@ -482,6 +540,9 @@ impl SessionInner { } } } + recv(fetch_metadata) -> request => { + handle_fetch_metadata(&*output_spool, request, &mut fetch_metadata); + } // make this select non-blocking so we spend most of our time parked // in poll diff --git a/libshpool/src/session_restore.rs b/libshpool/src/session_restore.rs index 94a24ea3..4e5f279d 100644 --- a/libshpool/src/session_restore.rs +++ b/libshpool/src/session_restore.rs @@ -56,6 +56,10 @@ pub trait SessionSpool { /// Process bytes from pty master. fn process(&mut self, bytes: &[u8]); + + /// The current window title, or None if it is unset or empty. The title is + /// not checked to be UTF-8. + fn window_title(&self) -> Option<&[u8]>; } /// A spool that doesn't do anything. @@ -69,6 +73,10 @@ impl SessionSpool for NullSpool { } fn process(&mut self, _: &[u8]) {} + + fn window_title(&self) -> Option<&[u8]> { + None + } } /// A spool that restores the last screenful of content using shpool_vt100. @@ -100,6 +108,11 @@ impl SessionSpool for Vt100Screen { fn process(&mut self, bytes: &[u8]) { self.parser.process(bytes) } + + fn window_title(&self) -> Option<&[u8]> { + Some(self.parser.screen().title().as_bytes()) + .filter(|window_title| !window_title.is_empty()) + } } /// A spool that restores the last n lines of content using shpool_vt100. @@ -130,6 +143,11 @@ impl SessionSpool for Vt100Lines { fn process(&mut self, bytes: &[u8]) { self.parser.process(bytes) } + + fn window_title(&self) -> Option<&[u8]> { + Some(self.parser.screen().title().as_bytes()) + .filter(|window_title| !window_title.is_empty()) + } } /// A spool that restores the last screenful of content using shpool-vterm. @@ -157,6 +175,10 @@ impl SessionSpool for Vterm { fn process(&mut self, bytes: &[u8]) { self.term.process(bytes); } + + fn window_title(&self) -> Option<&[u8]> { + self.term.title() + } } /// Creates a spool given a `mode`. diff --git a/shpool-protocol/src/lib.rs b/shpool-protocol/src/lib.rs index 0da25fc3..10f28264 100644 --- a/shpool-protocol/src/lib.rs +++ b/shpool-protocol/src/lib.rs @@ -294,6 +294,12 @@ pub struct Session { /// independently of `status`, so they can briefly disagree. #[serde(default)] pub attachments: Vec, + /// This session's window title. None if it is unset or empty, or + /// `session_restore_mode` is `simple`, or the session's shell has exited, + /// or the session was too busy to report it in time. Non-UTF-8 bytes are + /// replaced with U+FFFD. + #[serde(default)] + pub window_title: Option, } /// A session's attachment. diff --git a/shpool/tests/list.rs b/shpool/tests/list.rs index d275ec68..0eb4e4bd 100644 --- a/shpool/tests/list.rs +++ b/shpool/tests/list.rs @@ -1,4 +1,4 @@ -use std::process::Command; +use std::process::{Child, Command, Stdio}; use anyhow::{anyhow, Context}; use ntest::timeout; @@ -541,3 +541,149 @@ fn json_attachment_kill_removes_session() -> anyhow::Result<()> { Ok(()) } + +/// Attach a session named `name` and wait for the daemon to start streaming it. +fn attach_session( + daemon_proc: &mut support::daemon::Proc, + name: &str, +) -> anyhow::Result { + let bidi_enter_w = daemon_proc.events.take().unwrap().waiter(["daemon-bidi-stream-enter"]); + let sess = daemon_proc.attach(name, Default::default())?; + daemon_proc.events = Some(bidi_enter_w.wait_final_event("daemon-bidi-stream-enter")?); + Ok(sess) +} + +/// Start a daemon with `config` and attach a session named sh1 to it. +fn attach_sh1(config: &str) -> anyhow::Result<(support::daemon::Proc, support::attach::Proc)> { + let mut daemon_proc = support::daemon::Proc::new(config, DaemonArgs::default()) + .context("starting daemon proc")?; + let sess1 = attach_session(&mut daemon_proc, "sh1")?; + Ok((daemon_proc, sess1)) +} + +/// The window title `shpool list --json` reports for the session named `name`. +fn window_title(daemon_proc: &mut support::daemon::Proc, name: &str) -> anyhow::Result { + let sessions = list_sessions(daemon_proc)?; + let session = + sessions.iter().find(|s| s["name"] == name).ok_or_else(|| anyhow!("{name} not found"))?; + Ok(session["window_title"].clone()) +} + +/// Have sh1's shell print `printf_arg`, and wait for `shpool list --json` to +/// report `want` as its window title. +fn assert_window_title_reported(config: &str, printf_arg: &str, want: &str) -> anyhow::Result<()> { + let (mut daemon_proc, mut sess1) = attach_sh1(config)?; + assert!( + window_title(&mut daemon_proc, "sh1")?.is_null(), + "expected no window title before one is set" + ); + + sess1.run_cmd(&format!("printf '{printf_arg}'"))?; + support::wait_until(|| Ok(window_title(&mut daemon_proc, "sh1")? == want)) + .with_context(|| format!("window title should be reported as {want:?}"))?; + + Ok(()) +} + +#[test] +#[timeout(30000)] +fn json_window_title_vt100() -> anyhow::Result<()> { + assert_window_title_reported("restore_screen.toml", r"\033]2;hello\007", "hello") +} + +#[test] +#[timeout(30000)] +fn json_window_title_vterm() -> anyhow::Result<()> { + assert_window_title_reported("vterm_screen.toml", r"\033]2;hello\007", "hello") +} + +#[test] +#[timeout(30000)] +fn json_window_title_invalid_utf8_replaced() -> anyhow::Result<()> { + // vterm keeps a window title that is not UTF-8, where vt100 ignores it, so this + // shows the replacement. + assert_window_title_reported("vterm_screen.toml", r"\033]2;caf\351\007", "caf\u{fffd}") +} + +#[test] +#[timeout(30000)] +fn json_window_title_null_in_simple_restore_mode() -> anyhow::Result<()> { + let (mut daemon_proc, mut sess1) = attach_sh1("norc.toml")?; + let mut lm1 = sess1.line_matcher()?; + + sess1.run_cmd(r"printf '\033]2;hello\007'; echo done")?; + // The daemon handles output in order, so once `done` arrives it has also seen + // the window title. + lm1.scan_until_re("done$")?; + assert!( + window_title(&mut daemon_proc, "sh1")?.is_null(), + "simple restore mode keeps no window title" + ); + + Ok(()) +} + +#[test] +#[timeout(30000)] +fn json_window_title_concurrent_lists() -> anyhow::Result<()> { + let (mut daemon_proc, mut sess1) = attach_sh1("vterm_screen.toml")?; + sess1.run_cmd(r"printf '\033]2;hello\007'")?; + support::wait_until(|| Ok(window_title(&mut daemon_proc, "sh1")? == "hello"))?; + + // Each list asks every session for its window title, so a burst of them must + // all be answered before the deadline. + let lists: anyhow::Result> = (0..20) + .map(|_| { + Command::new(support::shpool_bin()?) + .arg("--socket") + .arg(&daemon_proc.socket_path) + .args(["list", "--json"]) + .stdout(Stdio::piped()) + .spawn() + .context("spawning list --json proc") + }) + .collect(); + for list in lists? { + let out = list.wait_with_output().context("waiting for list --json proc")?; + let parsed: Value = serde_json::from_slice(&out.stdout).context("parsing JSON output")?; + assert_eq!(parsed["sessions"][0]["window_title"], "hello"); + } + + Ok(()) +} + +#[test] +#[timeout(30000)] +fn json_window_title_null_when_session_does_not_answer() -> anyhow::Result<()> { + let mut daemon_proc = support::daemon::Proc::new("vterm_screen.toml", DaemonArgs::default()) + .context("starting daemon proc")?; + let mut sess2 = attach_session(&mut daemon_proc, "sh2")?; + sess2.run_cmd(r"printf '\033]2;two\007'")?; + support::wait_until(|| Ok(window_title(&mut daemon_proc, "sh2")? == "two"))?; + daemon_proc.detach(vec![String::from("sh2")])?; + let mut sess1 = attach_session(&mut daemon_proc, "sh1")?; + sess1.run_cmd(r"printf '\033]2;one\007'")?; + support::wait_until(|| Ok(window_title(&mut daemon_proc, "sh1")? == "one"))?; + + // Only sh1 is attached, so only its shell->client thread writes heartbeats, and + // pausing it there keeps it from answering. + daemon_proc.send_event_command("pause-at daemon-wrote-heartbeat")?; + daemon_proc.await_event("paused-at daemon-wrote-heartbeat")?; + let sessions = list_sessions(&mut daemon_proc)?; + daemon_proc.send_event_command("release daemon-wrote-heartbeat")?; + + let title = + |name: &str| sessions.iter().find(|s| s["name"] == name).map(|s| s["window_title"].clone()); + assert_eq!( + title("sh1"), + Some(Value::Null), + "sh1 cannot answer, so it is listed without a window title" + ); + assert_eq!( + title("sh2"), + Some(Value::from("two")), + "sh2 answers, so its window title is listed" + ); + + Ok(()) +}