diff --git a/crates/jcode-terminal-image/src/display.rs b/crates/jcode-terminal-image/src/display.rs index 3bf7eae58..ae5b4c065 100644 --- a/crates/jcode-terminal-image/src/display.rs +++ b/crates/jcode-terminal-image/src/display.rs @@ -1,6 +1,6 @@ //! Terminal image display support //! -//! Supports Kitty graphics protocol (Kitty, Ghostty), iTerm2 inline images, +//! Supports Kitty graphics protocol (Kitty, Ghostty, Handterm), iTerm2 inline images, //! and Sixel graphics (xterm, foot, mlterm, WezTerm). //! Falls back to a simple placeholder if no image protocol is available. @@ -40,16 +40,16 @@ impl ImageProtocol { return Self::Kitty; } - // Check TERM for kitty or ghostty + // Check TERM for Kitty-compatible terminals. if let Ok(term) = std::env::var("TERM") - && (term.contains("kitty") || term.contains("ghostty")) + && is_kitty_terminal_name(&term) { return Self::Kitty; } - // Check TERM_PROGRAM for Ghostty + // Check TERM_PROGRAM for Kitty-compatible terminals. if let Ok(term_program) = std::env::var("TERM_PROGRAM") { - if term_program == "ghostty" { + if is_kitty_terminal_name(&term_program) { return Self::Kitty; } if term_program == "iTerm.app" { @@ -113,6 +113,11 @@ impl ImageProtocol { } } +fn is_kitty_terminal_name(value: &str) -> bool { + let value = value.to_ascii_lowercase(); + value.contains("kitty") || value.contains("ghostty") || value.contains("handterm") +} + /// Display parameters for terminal images #[derive(Debug, Clone)] pub struct ImageDisplayParams { @@ -452,6 +457,12 @@ mod tests { assert!(!can_display_to_stdout(ImageProtocol::None, true)); } + #[test] + fn handterm_uses_kitty_graphics_protocol() { + assert!(is_kitty_terminal_name("handterm")); + assert!(is_kitty_terminal_name("HandTerm")); + } + #[test] fn test_calculate_display_size() { // Wide image diff --git a/crates/jcode-tui-mermaid/src/mermaid_runtime.rs b/crates/jcode-tui-mermaid/src/mermaid_runtime.rs index afcdb9dbc..72647d95a 100644 --- a/crates/jcode-tui-mermaid/src/mermaid_runtime.rs +++ b/crates/jcode-tui-mermaid/src/mermaid_runtime.rs @@ -141,6 +141,7 @@ pub(super) fn infer_protocol_from_env( || term_program.contains("kitty") || term_program.contains("wezterm") || term_program.contains("ghostty") + || term_program.contains("handterm") { return Some(ProtocolType::Kitty); } @@ -479,6 +480,10 @@ mod tests { infer_protocol_from_env(None, Some("ghostty"), None, None), Some(ProtocolType::Kitty) ); + assert_eq!( + infer_protocol_from_env(None, Some("HandTerm"), None, None), + Some(ProtocolType::Kitty) + ); assert_eq!( infer_protocol_from_env(None, Some("WezTerm"), None, None), Some(ProtocolType::Kitty) diff --git a/crates/jcode-tui/src/tui/app/handterm_native_scroll.rs b/crates/jcode-tui/src/tui/app/handterm_native_scroll.rs index ebccfdeb0..46977a365 100644 --- a/crates/jcode-tui/src/tui/app/handterm_native_scroll.rs +++ b/crates/jcode-tui/src/tui/app/handterm_native_scroll.rs @@ -15,7 +15,12 @@ use std::path::PathBuf; #[cfg(unix)] use std::sync::mpsc::{self, Receiver, Sender}; #[cfg(unix)] -use std::thread; +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; +#[cfg(unix)] +use std::thread::{self, JoinHandle}; #[cfg(unix)] use std::time::Duration; @@ -67,6 +72,10 @@ pub(super) struct HandtermNativeScrollClient { commands_rx: UnboundedReceiver, #[cfg(unix)] last_sent: Option, + #[cfg(unix)] + stop: Arc, + #[cfg(unix)] + thread: Option>, } impl HandtermNativeScrollClient { @@ -79,19 +88,22 @@ impl HandtermNativeScrollClient { #[cfg(unix)] { let socket_path = std::env::var_os(ENV_SOCKET).map(PathBuf::from)?; - Self::connect(socket_path).ok() + Self::connect(socket_path) } } #[cfg(unix)] - fn connect(socket_path: PathBuf) -> Result { + fn connect(socket_path: PathBuf) -> Option { let (updates_tx, updates_rx) = mpsc::channel(); let (commands_tx, commands_rx) = unbounded_channel(); - spawn_bridge_thread(socket_path, updates_rx, commands_tx); - Ok(Self { + let stop = Arc::new(AtomicBool::new(false)); + let thread = spawn_bridge_thread(socket_path, updates_rx, commands_tx, stop.clone())?; + Some(Self { updates_tx, commands_rx, last_sent: None, + stop, + thread: Some(thread), }) } @@ -108,10 +120,15 @@ impl HandtermNativeScrollClient { if self.last_sent.as_ref() == Some(&snapshot) { return; } - self.last_sent = Some(snapshot.clone()); - let _ = self.updates_tx.send(AppToHost::PaneSnapshot { - panes: snapshot.panes, - }); + if self + .updates_tx + .send(AppToHost::PaneSnapshot { + panes: snapshot.panes.clone(), + }) + .is_ok() + { + self.last_sent = Some(snapshot); + } } } @@ -120,6 +137,16 @@ impl HandtermNativeScrollClient { } } +#[cfg(unix)] +impl Drop for HandtermNativeScrollClient { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + impl App { #[cfg(unix)] fn current_native_scroll_snapshot(&self) -> PaneSnapshot { @@ -182,17 +209,20 @@ fn spawn_bridge_thread( socket_path: PathBuf, updates_rx: Receiver, commands_tx: UnboundedSender, -) { - if let Err(err) = thread::Builder::new() + stop: Arc, +) -> Option> { + match thread::Builder::new() .name("jcode-handterm-scroll".to_string()) - .spawn(move || { - let _ = bridge_thread(socket_path, updates_rx, commands_tx); - }) + .spawn(move || bridge_thread(socket_path, updates_rx, commands_tx, stop)) { - crate::logging::warn(&format!( - "Failed to spawn handterm native scroll bridge thread: {}", - err - )); + Ok(thread) => Some(thread), + Err(err) => { + crate::logging::warn(&format!( + "Failed to spawn handterm native scroll bridge thread: {}", + err + )); + None + } } } @@ -201,72 +231,95 @@ fn bridge_thread( socket_path: PathBuf, updates_rx: Receiver, commands_tx: UnboundedSender, -) -> Result<()> { - let mut stream = connect_with_retry(&socket_path)?; - stream - .set_nonblocking(true) - .context("failed setting native scroll socket nonblocking")?; - let mut read_buf = Vec::new(); - - loop { + stop: Arc, +) { + let mut latest_update = None::; + while !stop.load(Ordering::Relaxed) { while let Ok(update) = updates_rx.try_recv() { - write_line(&mut stream, &update)?; + latest_update = Some(update); } - - let mut chunk = [0u8; 4096]; - loop { - match stream.read(&mut chunk) { - Ok(0) => return Ok(()), - Ok(n) => read_buf.extend_from_slice(&chunk[..n]), - Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break, - Err(err) => return Err(err).context("failed reading native scroll command"), - } + let Some(mut stream) = connect_with_retry(&socket_path, &stop) else { + break; + }; + if stream.set_nonblocking(true).is_err() { + continue; } - - while let Some(pos) = read_buf.iter().position(|&b| b == b'\n') { - let line = read_buf.drain(..=pos).collect::>(); - let line = &line[..line.len().saturating_sub(1)]; - if line.is_empty() { - continue; + while let Ok(update) = updates_rx.try_recv() { + latest_update = Some(update); + } + if let Some(update) = latest_update.as_ref() + && write_line(&mut stream, update).is_err() + { + continue; + } + let mut read_buf = Vec::new(); + while !stop.load(Ordering::Relaxed) { + let mut disconnected = false; + let mut update_pending = false; + while let Ok(update) = updates_rx.try_recv() { + latest_update = Some(update); + update_pending = true; + } + if update_pending + && let Some(update) = latest_update.as_ref() + && write_line(&mut stream, update).is_err() + { + disconnected = true; + } + if disconnected { + break; + } + let mut chunk = [0u8; 4096]; + loop { + match stream.read(&mut chunk) { + Ok(0) => { + disconnected = true; + break; + } + Ok(n) => read_buf.extend_from_slice(&chunk[..n]), + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break, + Err(_) => { + disconnected = true; + break; + } + } + } + if disconnected { + break; + } + while let Some(pos) = read_buf.iter().position(|&b| b == b'\n') { + let line = read_buf.drain(..=pos).collect::>(); + let line = &line[..line.len().saturating_sub(1)]; + if line.is_empty() { + continue; + } + match serde_json::from_slice::(line) { + Ok(command) => { + let _ = commands_tx.send(command); + } + Err(_) => { + disconnected = true; + break; + } + } + } + if disconnected { + break; } - let command = serde_json::from_slice::(line) - .context("failed decoding native scroll command")?; - let _ = commands_tx.send(command); + thread::sleep(Duration::from_millis(8)); } - - thread::sleep(Duration::from_millis(8)); } } #[cfg(unix)] -fn connect_with_retry(socket_path: &PathBuf) -> Result { - let deadline = std::time::Instant::now() + Duration::from_secs(10); - loop { +fn connect_with_retry(socket_path: &PathBuf, stop: &AtomicBool) -> Option { + while !stop.load(Ordering::Relaxed) { match UnixStream::connect(socket_path) { - Ok(stream) => return Ok(stream), - Err(err) if std::time::Instant::now() < deadline => { - if err.kind() != std::io::ErrorKind::NotFound - && err.kind() != std::io::ErrorKind::ConnectionRefused - { - return Err(err).with_context(|| { - format!( - "failed connecting handterm native scroll socket {}", - socket_path.display() - ) - }); - } - thread::sleep(Duration::from_millis(50)); - } - Err(err) => { - return Err(err).with_context(|| { - format!( - "failed connecting handterm native scroll socket {}", - socket_path.display() - ) - }); - } + Ok(stream) => return Some(stream), + Err(_) => thread::sleep(Duration::from_millis(50)), } } + None } #[cfg(unix)] @@ -277,3 +330,127 @@ fn write_line(stream: &mut UnixStream, message: &T) -> Result<()> .write_all(&bytes) .context("failed writing native scroll state") } + +#[cfg(all(test, unix))] +mod bridge_tests { + use super::*; + use std::io::{BufRead, BufReader}; + use std::os::unix::net::UnixListener; + use std::time::{Instant, SystemTime, UNIX_EPOCH}; + + fn unique_socket_path(label: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should be after epoch") + .as_nanos(); + std::env::temp_dir().join(format!( + "jcode-handterm-{label}-{}-{nonce:x}.sock", + std::process::id() + )) + } + + fn accept_with_timeout(listener: &UnixListener) -> UnixStream { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + match listener.accept() { + Ok((stream, _)) => return stream, + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "native scroll client did not connect" + ); + thread::sleep(Duration::from_millis(10)); + } + Err(err) => panic!("failed accepting native scroll client: {err}"), + } + } + } + + fn snapshot(position: usize) -> AppToHost { + AppToHost::PaneSnapshot { + panes: vec![PaneState { + kind: PaneKind::Chat, + x: 1, + y: 2, + width: 30, + height: 12, + position, + content_length: 100, + viewport_length: 12, + }], + } + } + + fn read_snapshot(stream: &UnixStream) -> AppToHost { + stream + .set_nonblocking(false) + .expect("accepted stream should become blocking"); + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .expect("read timeout should set"); + let mut line = String::new(); + BufReader::new(stream) + .read_line(&mut line) + .expect("snapshot line should arrive"); + serde_json::from_str(line.trim_end()).expect("snapshot should decode") + } + + #[test] + fn bridge_reconnects_and_replays_latest_snapshot() { + let socket_path = unique_socket_path("reconnect"); + let listener = UnixListener::bind(&socket_path).expect("test socket should bind"); + listener + .set_nonblocking(true) + .expect("listener should become nonblocking"); + let (updates_tx, updates_rx) = mpsc::channel(); + let (commands_tx, _commands_rx) = unbounded_channel(); + let stop = Arc::new(AtomicBool::new(false)); + let thread = + spawn_bridge_thread(socket_path.clone(), updates_rx, commands_tx, stop.clone()) + .expect("bridge thread should spawn"); + + let first = accept_with_timeout(&listener); + let latest = snapshot(7); + updates_tx + .send(latest.clone()) + .expect("snapshot should queue"); + assert_eq!(read_snapshot(&first), latest); + drop(first); + + let second = accept_with_timeout(&listener); + assert_eq!(read_snapshot(&second), latest); + + stop.store(true, Ordering::Relaxed); + thread.join().expect("bridge thread should stop cleanly"); + let _ = std::fs::remove_file(socket_path); + } + + #[test] + fn bridge_coalesces_updates_while_waiting_for_host() { + let socket_path = unique_socket_path("coalesce"); + let (updates_tx, updates_rx) = mpsc::channel(); + let (commands_tx, _commands_rx) = unbounded_channel(); + let stop = Arc::new(AtomicBool::new(false)); + let thread = + spawn_bridge_thread(socket_path.clone(), updates_rx, commands_tx, stop.clone()) + .expect("bridge thread should spawn"); + + updates_tx + .send(snapshot(1)) + .expect("old snapshot should queue"); + let latest = snapshot(9); + updates_tx + .send(latest.clone()) + .expect("latest snapshot should queue"); + let listener = UnixListener::bind(&socket_path).expect("test socket should bind"); + listener + .set_nonblocking(true) + .expect("listener should become nonblocking"); + let stream = accept_with_timeout(&listener); + assert_eq!(read_snapshot(&stream), latest); + + stop.store(true, Ordering::Relaxed); + thread.join().expect("bridge thread should stop cleanly"); + let _ = std::fs::remove_file(socket_path); + } +} diff --git a/src/cli/commands/menubar.rs b/src/cli/commands/menubar.rs index b94e1fe1a..760f318d4 100644 --- a/src/cli/commands/menubar.rs +++ b/src/cli/commands/menubar.rs @@ -9,7 +9,7 @@ use anyhow::Result; use serde::Serialize; -use crate::session::{self, SessionCounts}; +use crate::session::{self, SessionCounts, SessionPresence}; #[derive(Debug, Serialize)] struct CountsReport { @@ -26,6 +26,32 @@ impl From for CountsReport { } } +/// Return whether a persisted session represents a top-level session opened by +/// the user. The shared server also tracks internal debug/swarm runtimes in the +/// active-PID registry, but those are implementation details rather than menu +/// entries. +fn load_user_root_session(session_id: &str) -> Option { + session::Session::load_startup_stub(session_id) + .ok() + .map(|session| session.parent_id.is_none() && !session.is_debug) +} + +fn user_root_session_presence() -> Vec { + session::user_session_presence() + .into_iter() + // A marker can briefly precede its first persisted snapshot. Keep an + // unknown session visible for this read rather than hiding a new window. + .filter(|presence| load_user_root_session(&presence.session_id).unwrap_or(true)) + .collect() +} + +fn counts_for_presence(sessions: &[SessionPresence]) -> SessionCounts { + SessionCounts { + total: sessions.len(), + streaming: sessions.iter().filter(|session| session.streaming).count(), + } +} + /// Format the compact title shown next to the menu bar icon. /// /// Always shows both the streaming and total counts directly in the menu bar @@ -77,13 +103,16 @@ fn format_session_menu_item_title_with_display( pub fn run_menubar_command(once: bool, json: bool) -> Result<()> { if json { - let report = CountsReport::from(session::user_session_counts()); + let report = CountsReport::from(counts_for_presence(&user_root_session_presence())); println!("{}", serde_json::to_string(&report)?); return Ok(()); } if once { - println!("{}", format_menubar_summary(session::user_session_counts())); + println!( + "{}", + format_menubar_summary(counts_for_presence(&user_root_session_presence())) + ); return Ok(()); } @@ -251,11 +280,15 @@ fn is_menubar_sandbox( #[cfg(target_os = "macos")] mod macos { - use super::{format_menubar_summary, format_menubar_title, format_session_menu_item_title}; - use crate::session::{self, SessionCounts, SessionPresence}; + use super::{ + counts_for_presence, format_menubar_summary, format_menubar_title, + format_session_menu_item_title, load_user_root_session, + }; + use crate::session::{self, SessionPresence}; use std::cell::RefCell; use std::cmp::Reverse; + use std::collections::{HashMap, HashSet}; use objc2::rc::Retained; use objc2::runtime::AnyObject; @@ -524,6 +557,10 @@ mod macos { status_item.setMenu(Some(&menu)); let last_sessions: RefCell> = RefCell::new(Vec::new()); + // Classification does not change for a session ID. Cache it so a large + // population of internal workers costs one metadata read each rather + // than one read per worker every second. + let root_visibility: RefCell> = RefCell::new(HashMap::new()); let app_for_refresh = app.clone(); let refresh = move || { // Re-sync the Light/Dark appearance each tick so toggling the @@ -532,12 +569,29 @@ mod macos { sync_app_appearance(&app_for_refresh); let mut sessions = session::user_session_presence(); + let active_ids: HashSet = sessions + .iter() + .map(|presence| presence.session_id.clone()) + .collect(); + root_visibility + .borrow_mut() + .retain(|session_id, _| active_ids.contains(session_id)); + sessions.retain(|presence| { + if let Some(visible) = root_visibility.borrow().get(&presence.session_id).copied() { + return visible; + } + let Some(visible) = load_user_root_session(&presence.session_id) else { + // The marker may race the first session save. Retry next tick. + return true; + }; + root_visibility + .borrow_mut() + .insert(presence.session_id.clone(), visible); + visible + }); sessions.sort_by_key(|s| (Reverse(s.streaming), s.session_id.clone())); - let counts = SessionCounts { - total: sessions.len(), - streaming: sessions.iter().filter(|s| s.streaming).count(), - }; + let counts = counts_for_presence(&sessions); if let Some(button) = status_item.button(mtm) { let title = format_menubar_title(counts); let attributed = attributed_title(&title, &title_font, counts.streaming > 0); @@ -833,6 +887,58 @@ mod tests { } } + #[test] + fn menu_presence_includes_only_user_root_sessions() { + let _guard = crate::storage::lock_test_env(); + let previous_home = std::env::var_os("JCODE_HOME"); + let temp = tempfile::tempdir().expect("create temporary JCODE_HOME"); + crate::env::set_var("JCODE_HOME", temp.path()); + + let mut root = crate::session::Session::create_with_id( + "session_fox_root".to_string(), + None, + Some("Root work".to_string()), + ); + root.mark_active(); + root.save().expect("save root session"); + + let mut debug = crate::session::Session::create_with_id( + "session_owl_debug".to_string(), + None, + Some("Internal worker".to_string()), + ); + debug.set_debug(true); + debug.mark_active(); + debug.save().expect("save debug session"); + + let mut child = crate::session::Session::create_with_id( + "session_bear_child".to_string(), + Some(root.id.clone()), + Some("Child work".to_string()), + ); + child.mark_active(); + child.save().expect("save child session"); + + let visible = user_root_session_presence(); + assert_eq!( + visible + .iter() + .map(|presence| presence.session_id.as_str()) + .collect::>(), + vec!["session_fox_root"] + ); + assert_eq!(counts_for_presence(&visible).total, 1); + + root.mark_closed(); + debug.mark_closed(); + child.mark_closed(); + if let Some(previous_home) = previous_home { + crate::env::set_var("JCODE_HOME", previous_home); + } else { + crate::env::remove_var("JCODE_HOME"); + } + } + #[test] fn counts_report_serializes_to_json() { let report = CountsReport::from(SessionCounts {