terminal works basically. (#12189)

* terminal works basically.
todo:
- persistent
- sessions restore
- web
- mobile

* missed terminal persistent option change

* android sdk 34 -> 35

* +#![cfg_attr(lt_1_77, feature(c_str_literals))]

* fixing ci

* fix ci

* fix ci for android

* try "Fix Android SDK Platform 35"

* fix android 34

* revert flutter_plugin_android_lifecycle to 2.0.17 which used in rustdesk 1.4.0

* refactor, but break something of desktop terminal (new tab showing loading)

* fix connecting...

src/server/connection.rs

@@ -169,6 +169,7 @@ pub enum AuthConnType {
     FileTransfer,
     PortForward,
     ViewCamera,
+    Terminal,
 }
 
 pub struct Connection {
@@ -182,6 +183,7 @@ pub struct Connection {
     file_timer: crate::RustDeskInterval,
     file_transfer: Option<(String, bool)>,
     view_camera: bool,
+    terminal: bool,
     port_forward_socket: Option<Framed<TcpStream, BytesCodec>>,
     port_forward_address: String,
     tx_to_cm: mpsc::UnboundedSender<ipc::Data>,
@@ -250,6 +252,9 @@ pub struct Connection {
     // For post requests that need to be sent sequentially.
     // eg. post_conn_audit
     tx_post_seq: mpsc::UnboundedSender<(String, Value)>,
+    terminal_service_id: String,
+    terminal_persistent: bool,
+    terminal_generic_service: Option<Box<GenericService>>,
 }
 
 impl ConnInner {
@@ -347,6 +352,7 @@ impl Connection {
             file_timer: crate::rustdesk_interval(time::interval(SEC30)),
             file_transfer: None,
             view_camera: false,
+            terminal: false,
             port_forward_socket: None,
             port_forward_address: "".to_owned(),
             tx_to_cm,
@@ -410,6 +416,9 @@ impl Connection {
             tx_from_authed,
             printer_data: Vec::new(),
             tx_post_seq,
+            terminal_service_id: "".to_owned(),
+            terminal_persistent: false,
+            terminal_generic_service: None,
         };
         let addr = hbb_common::try_into_v4(addr);
         if !conn.on_open(addr).await {
@@ -450,7 +459,7 @@ impl Connection {
         let mut last_recv_time = Instant::now();
 
         conn.stream.set_send_timeout(
-            if conn.file_transfer.is_some() || conn.port_forward_socket.is_some() {
+            if conn.file_transfer.is_some() || conn.port_forward_socket.is_some() || conn.terminal {
                 SEND_TIMEOUT_OTHER
             } else {
                 SEND_TIMEOUT_VIDEO
@@ -1253,6 +1262,8 @@ impl Connection {
             (2, AuthConnType::PortForward)
         } else if self.view_camera {
             (3, AuthConnType::ViewCamera)
+        } else if self.terminal {
+            (4, AuthConnType::Terminal)
         } else {
             (0, AuthConnType::Remote)
         };
@@ -1361,8 +1372,7 @@ impl Connection {
             return;
         }
         #[cfg(target_os = "linux")]
-        if !self.file_transfer.is_some() && !self.port_forward_socket.is_some() && !self.view_camera
-        {
+        if self.is_remote() {
             let mut msg = "".to_string();
             if crate::platform::linux::is_login_screen_wayland() {
                 msg = crate::client::LOGIN_SCREEN_WAYLAND.to_owned()
@@ -1393,7 +1403,8 @@ impl Connection {
         }
         #[cfg(not(any(target_os = "android", target_os = "ios")))]
         if self.file_transfer.is_some() {
-            if crate::platform::is_prelogin() { // }|| self.tx_to_cm.send(ipc::Data::Test).is_err() {
+            if crate::platform::is_prelogin() {
+                // }|| self.tx_to_cm.send(ipc::Data::Test).is_err() {
                 username = "".to_owned();
             }
         }
@@ -1408,6 +1419,8 @@ impl Connection {
         pi.sas_enabled = sas_enabled;
         pi.features = Some(Features {
             privacy_mode: privacy_mode::is_privacy_mode_supported(),
+            #[cfg(not(any(target_os = "android", target_os = "ios")))]
+            terminal: true, // Terminal feature is supported on desktop only
             ..Default::default()
         })
         .into();
@@ -1417,7 +1430,7 @@ impl Connection {
         let mut wait_session_id_confirm = false;
         #[cfg(windows)]
         self.handle_windows_specific_session(&mut pi, &mut wait_session_id_confirm);
-        if self.file_transfer.is_some() {
+        if self.file_transfer.is_some() || self.terminal {
             res.set_peer_info(pi);
         } else if self.view_camera {
             let supported_encoding = scrap::codec::Encoder::supported_encoding();
@@ -1509,6 +1522,10 @@ impl Connection {
             } else {
                 self.delayed_read_dir = Some((dir.to_owned(), show_hidden));
             }
+        } else if self.terminal {
+            self.keyboard = false;
+            #[cfg(not(any(target_os = "android", target_os = "ios")))]
+            self.init_terminal_service().await;
         } else if self.view_camera {
             if !wait_session_id_confirm {
                 self.try_sub_camera_displays();
@@ -1531,9 +1548,16 @@ impl Connection {
         }
     }
 
+    #[inline]
+    fn is_remote(&self) -> bool {
+        self.file_transfer.is_none()
+            && self.port_forward_socket.is_none()
+            && !self.view_camera
+            && !self.terminal
+    }
+
     fn try_sub_monitor_services(&mut self) {
-        let is_remote =
-            self.file_transfer.is_none() && self.port_forward_socket.is_none() && !self.view_camera;
+        let is_remote = self.is_remote();
         if is_remote && !self.services_subed {
             self.services_subed = true;
             if let Some(s) = self.server.upgrade() {
@@ -1651,6 +1675,7 @@ impl Connection {
             id: self.inner.id(),
             is_file_transfer: self.file_transfer.is_some(),
             is_view_camera: self.view_camera,
+            is_terminal: self.terminal,
             port_forward: self.port_forward_address.clone(),
             peer_id,
             name,
@@ -1902,6 +1927,19 @@ impl Connection {
                     }
                     self.view_camera = true;
                 }
+                Some(login_request::Union::Terminal(terminal)) => {
+                    if !Connection::permission(keys::OPTION_ENABLE_TERMINAL) {
+                        self.send_login_error("No permission of terminal").await;
+                        sleep(1.).await;
+                        return false;
+                    }
+                    self.terminal = true;
+                    if let Some(o) = self.options_in_login.as_ref() {
+                        self.terminal_persistent =
+                            o.terminal_persistent.enum_value() == Ok(BoolOption::Yes);
+                    }
+                    self.terminal_service_id = terminal.service_id;
+                }
                 Some(login_request::Union::PortForward(mut pf)) => {
                     if !Connection::permission("enable-tunnel") {
                         self.send_login_error("No permission of IP tunneling").await;
@@ -2791,7 +2829,7 @@ impl Connection {
                                 }
                             } else if self.view_camera {
                                 self.try_sub_camera_displays();
-                            } else {
+                            } else if !self.terminal {
                                 self.try_sub_monitor_services();
                             }
                         }
@@ -2843,6 +2881,12 @@ impl Connection {
                         self.refresh_video_display(Some(request.display as usize));
                     }
                 }
+                Some(message::Union::TerminalAction(action)) => {
+                    #[cfg(not(any(target_os = "android", target_os = "ios")))]
+                    allow_err!(self.handle_terminal_action(action).await);
+                    #[cfg(any(target_os = "android", target_os = "ios"))]
+                    log::warn!("Terminal action received but not supported on this platform");
+                }
                 _ => {}
             }
         }
@@ -3371,6 +3415,12 @@ impl Connection {
                 }
             }
         }
+        #[cfg(not(any(target_os = "android", target_os = "ios")))]
+        if let Ok(q) = o.terminal_persistent.enum_value() {
+            if q != BoolOption::NotSet {
+                self.update_terminal_persistence(q == BoolOption::Yes).await;
+            }
+        }
     }
 
     async fn turn_on_privacy(&mut self, impl_key: String) {
@@ -3562,12 +3612,7 @@ impl Connection {
 
     #[cfg(windows)]
     fn portable_check(&mut self) {
-        if self.portable.is_installed
-            || self.file_transfer.is_some()
-            || self.view_camera
-            || self.port_forward_socket.is_some()
-            || !self.keyboard
-        {
+        if self.portable.is_installed || !self.is_remote() || !self.keyboard {
             return;
         }
         let running = portable_client::running();
@@ -3779,6 +3824,55 @@ impl Connection {
         msg_out.set_message_box(res);
         self.send(msg_out).await;
     }
+
+    #[cfg(not(any(target_os = "android", target_os = "ios")))]
+    async fn update_terminal_persistence(&mut self, persistent: bool) {
+        self.terminal_persistent = persistent;
+        terminal_service::set_persistent(&self.terminal_service_id, persistent).ok();
+    }
+
+    #[cfg(not(any(target_os = "android", target_os = "ios")))]
+    async fn init_terminal_service(&mut self) {
+        if self.terminal_service_id.is_empty() {
+            self.terminal_service_id = terminal_service::generate_service_id();
+        }
+        let s = Box::new(terminal_service::new(
+            self.terminal_service_id.clone(),
+            self.terminal_persistent,
+        ));
+        s.on_subscribe(self.inner.clone());
+        self.terminal_generic_service = Some(s);
+    }
+
+    #[cfg(not(any(target_os = "android", target_os = "ios")))]
+    async fn handle_terminal_action(&mut self, action: TerminalAction) -> ResultType<()> {
+        let mut proxy = terminal_service::TerminalServiceProxy::new(
+            self.terminal_service_id.clone(),
+            Some(self.terminal_persistent),
+        );
+
+        match proxy.handle_action(&action) {
+            Ok(Some(response)) => {
+                let mut msg_out = Message::new();
+                msg_out.set_terminal_response(response);
+                self.send(msg_out).await;
+            }
+            Ok(None) => {
+                // No response needed
+            }
+            Err(err) => {
+                let mut response = TerminalResponse::new();
+                let mut error = TerminalError::new();
+                error.message = format!("Failed to handle action: {}", err);
+                response.set_error(error);
+                let mut msg_out = Message::new();
+                msg_out.set_terminal_response(response);
+                self.send(msg_out).await;
+            }
+        }
+
+        Ok(())
+    }
 }
 
 pub fn insert_switch_sides_uuid(id: String, uuid: uuid::Uuid) {
@@ -4151,6 +4245,10 @@ impl Drop for Connection {
     fn drop(&mut self) {
         #[cfg(not(any(target_os = "android", target_os = "ios")))]
         self.release_pressed_modifiers();
+
+        if let Some(s) = self.terminal_generic_service.as_ref() {
+            s.join();
+        }
     }
 }
 

src/server/terminal_service.rs

@@ -0,0 +1,968 @@
+use super::*;
+use hbb_common::{
+    anyhow::{anyhow, Context, Result},
+    compress,
+};
+use portable_pty::{Child, CommandBuilder, PtySize};
+use std::{
+    collections::{HashMap, VecDeque},
+    io::{Read, Write},
+    sync::{
+        mpsc::{self, Receiver, SyncSender},
+        Arc, Mutex,
+    },
+    thread,
+    time::{Duration, Instant},
+};
+
+const MAX_OUTPUT_BUFFER_SIZE: usize = 1024 * 1024; // 1MB per terminal
+const MAX_BUFFER_LINES: usize = 10000;
+const MAX_SERVICES: usize = 100; // Maximum number of persistent terminal services
+const SERVICE_IDLE_TIMEOUT: Duration = Duration::from_secs(3600); // 1 hour idle timeout
+const CHANNEL_BUFFER_SIZE: usize = 100; // Number of messages to buffer in channel
+const COMPRESS_THRESHOLD: usize = 512; // Compress terminal data larger than this
+
+lazy_static::lazy_static! {
+    // Global registry of persistent terminal services indexed by service_id
+    static ref TERMINAL_SERVICES: Arc<Mutex<HashMap<String, Arc<Mutex<PersistentTerminalService>>>>> =
+        Arc::new(Mutex::new(HashMap::new()));
+
+    // Cleanup task handle
+    static ref CLEANUP_TASK: Arc<Mutex<Option<std::thread::JoinHandle<()>>>> = Arc::new(Mutex::new(None));
+
+    // List of terminal child processes to check for zombies
+    static ref TERMINAL_TASKS: Arc<Mutex<Vec<Box<dyn Child + Send + Sync>>>> = Arc::new(Mutex::new(Vec::new()));
+}
+
+/// Service metadata that is sent to clients
+#[derive(Clone, Debug)]
+pub struct ServiceMetadata {
+    pub service_id: String,
+    pub created_at: Instant,
+    pub terminal_count: usize,
+    pub is_persistent: bool,
+}
+
+/// Generate a new persistent service ID
+pub fn generate_service_id() -> String {
+    format!("ts_{}", uuid::Uuid::new_v4())
+}
+
+fn get_default_shell() -> String {
+    #[cfg(target_os = "windows")]
+    {
+        // Try PowerShell Core first (cross-platform version)
+        // Common installation paths for PowerShell Core
+        let pwsh_paths = [
+            "pwsh.exe",
+            r"C:\Program Files\PowerShell\7\pwsh.exe",
+            r"C:\Program Files\PowerShell\6\pwsh.exe",
+        ];
+
+        for path in &pwsh_paths {
+            if std::path::Path::new(path).exists() {
+                return path.to_string();
+            }
+        }
+
+        // Try Windows PowerShell (should be available on all Windows systems)
+        let powershell_path = r"C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe";
+        if std::path::Path::new(powershell_path).exists() {
+            return powershell_path.to_string();
+        }
+
+        // Final fallback to cmd.exe
+        std::env::var("COMSPEC").unwrap_or_else(|_| "cmd.exe".to_string())
+    }
+    #[cfg(not(target_os = "windows"))]
+    {
+        // First try the SHELL environment variable
+        if let Ok(shell) = std::env::var("SHELL") {
+            if !shell.is_empty() {
+                return shell;
+            }
+        }
+
+        // Check for common shells in order of preference
+        let shells = ["/bin/bash", "/bin/zsh", "/bin/sh"];
+        for shell in &shells {
+            if std::path::Path::new(shell).exists() {
+                return shell.to_string();
+            }
+        }
+
+        // Final fallback to /bin/sh which should exist on all POSIX systems
+        "/bin/sh".to_string()
+    }
+}
+
+/// Get or create a persistent terminal service
+fn get_or_create_service(
+    service_id: String,
+    is_persistent: bool,
+) -> Result<Arc<Mutex<PersistentTerminalService>>> {
+    let mut services = TERMINAL_SERVICES.lock().unwrap();
+
+    // Check service limit
+    if !services.contains_key(&service_id) && services.len() >= MAX_SERVICES {
+        return Err(anyhow!(
+            "Maximum number of terminal services ({}) reached",
+            MAX_SERVICES
+        ));
+    }
+
+    let service = services
+        .entry(service_id.clone())
+        .or_insert_with(|| {
+            log::info!(
+                "Creating new terminal service: {} (persistent: {})",
+                service_id,
+                is_persistent
+            );
+            Arc::new(Mutex::new(PersistentTerminalService::new(
+                service_id.clone(),
+                is_persistent,
+            )))
+        })
+        .clone();
+
+    // Ensure cleanup task is running
+    ensure_cleanup_task();
+
+    Ok(service)
+}
+
+/// Remove a service from the global registry
+fn remove_service(service_id: &str) {
+    let mut services = TERMINAL_SERVICES.lock().unwrap();
+    if let Some(service) = services.remove(service_id) {
+        log::info!("Removed service: {}", service_id);
+        // Close all terminals in the service
+        let sessions = service.lock().unwrap().sessions.clone();
+        for (_, session) in sessions.iter() {
+            let mut session = session.lock().unwrap();
+            if let Some(mut child) = session.child.take() {
+                // Kill the process
+                let _ = child.kill();
+                add_to_reaper(child);
+            }
+        }
+    }
+}
+
+/// List all active terminal services
+pub fn list_services() -> Vec<ServiceMetadata> {
+    let services = TERMINAL_SERVICES.lock().unwrap();
+    services
+        .iter()
+        .filter_map(|(id, service)| {
+            service.lock().ok().map(|svc| ServiceMetadata {
+                service_id: id.clone(),
+                created_at: svc.created_at,
+                terminal_count: svc.sessions.len(),
+                is_persistent: svc.is_persistent,
+            })
+        })
+        .collect()
+}
+
+/// Get service by ID
+pub fn get_service(service_id: &str) -> Option<Arc<Mutex<PersistentTerminalService>>> {
+    let services = TERMINAL_SERVICES.lock().unwrap();
+    services.get(service_id).cloned()
+}
+
+/// Clean up inactive services
+pub fn cleanup_inactive_services() {
+    let services = TERMINAL_SERVICES.lock().unwrap();
+    let now = Instant::now();
+    let mut to_remove = Vec::new();
+
+    for (service_id, service) in services.iter() {
+        if let Ok(svc) = service.lock() {
+            // Remove non-persistent services after idle timeout
+            if !svc.is_persistent && now.duration_since(svc.last_activity) > SERVICE_IDLE_TIMEOUT {
+                to_remove.push(service_id.clone());
+                log::info!("Cleaning up idle non-persistent service: {}", service_id);
+            }
+            // Remove persistent services with no active terminals after longer timeout
+            else if svc.is_persistent
+                && svc.sessions.is_empty()
+                && now.duration_since(svc.last_activity) > SERVICE_IDLE_TIMEOUT * 2
+            {
+                to_remove.push(service_id.clone());
+                log::info!("Cleaning up empty persistent service: {}", service_id);
+            }
+        }
+    }
+
+    // Remove outside of iteration to avoid deadlock
+    drop(services);
+    for id in to_remove {
+        remove_service(&id);
+    }
+}
+
+/// Add a child process to the zombie reaper
+fn add_to_reaper(child: Box<dyn Child + Send + Sync>) {
+    if let Ok(mut tasks) = TERMINAL_TASKS.lock() {
+        tasks.push(child);
+    }
+}
+
+/// Check and reap zombie terminal processes
+fn check_zombie_terminals() {
+    let mut tasks = match TERMINAL_TASKS.lock() {
+        Ok(t) => t,
+        Err(_) => return,
+    };
+
+    let mut i = 0;
+    while i < tasks.len() {
+        match tasks[i].try_wait() {
+            Ok(Some(_)) => {
+                // Process has exited, remove it
+                log::info!("Process exited: {:?}", tasks[i].process_id());
+                tasks.remove(i);
+            }
+            Ok(None) => {
+                // Still running
+                i += 1;
+            }
+            Err(err) => {
+                // Error checking status, remove it
+                log::info!(
+                    "Process exited with error: {:?}, err: {err}",
+                    tasks[i].process_id()
+                );
+                tasks.remove(i);
+            }
+        }
+    }
+}
+
+/// Ensure the cleanup task is running
+fn ensure_cleanup_task() {
+    let mut task_handle = CLEANUP_TASK.lock().unwrap();
+    if task_handle.is_none() {
+        let handle = std::thread::spawn(|| {
+            log::info!("Started cleanup task");
+            let mut last_service_cleanup = Instant::now();
+            loop {
+                // Check for zombie processes every 100ms
+                check_zombie_terminals();
+
+                // Check for inactive services every 5 minutes
+                if last_service_cleanup.elapsed() > Duration::from_secs(300) {
+                    cleanup_inactive_services();
+                    last_service_cleanup = Instant::now();
+                }
+
+                std::thread::sleep(Duration::from_millis(100));
+            }
+        });
+        *task_handle = Some(handle);
+    }
+}
+
+pub fn new(service_id: String, is_persistent: bool) -> GenericService {
+    // Create the service with initial persistence setting
+    allow_err!(get_or_create_service(service_id.clone(), is_persistent));
+    let svc = EmptyExtraFieldService::new(service_id.clone(), false);
+    GenericService::run(&svc.clone(), move |sp| run(sp, service_id.clone()));
+    svc.sp
+}
+
+fn run(sp: EmptyExtraFieldService, service_id: String) -> ResultType<()> {
+    while sp.ok() {
+        let responses = TerminalServiceProxy::new(service_id.clone(), None).read_outputs();
+        for response in responses {
+            let mut msg_out = Message::new();
+            msg_out.set_terminal_response(response);
+            sp.send(msg_out);
+        }
+
+        thread::sleep(Duration::from_millis(30)); // Read at ~33fps for responsive terminal
+    }
+
+    // Clean up non-persistent service when loop exits
+    if let Some(service) = get_service(&service_id) {
+        let should_remove = !service.lock().unwrap().is_persistent;
+        if should_remove {
+            remove_service(&service_id);
+        }
+    }
+
+    Ok(())
+}
+
+/// Output buffer for terminal session
+struct OutputBuffer {
+    lines: VecDeque<Vec<u8>>,
+    total_size: usize,
+    last_line_incomplete: bool,
+}
+
+impl OutputBuffer {
+    fn new() -> Self {
+        Self {
+            lines: VecDeque::new(),
+            total_size: 0,
+            last_line_incomplete: false,
+        }
+    }
+
+    fn append(&mut self, data: &[u8]) {
+        if data.is_empty() {
+            return;
+        }
+
+        // Handle incomplete lines
+        let mut start = 0;
+        if self.last_line_incomplete {
+            if let Some(last_line) = self.lines.back_mut() {
+                // Find first newline in new data
+                if let Some(newline_pos) = data.iter().position(|&b| b == b'\n') {
+                    last_line.extend_from_slice(&data[..=newline_pos]);
+                    start = newline_pos + 1;
+                    self.last_line_incomplete = false;
+                } else {
+                    // Still no newline, append all
+                    last_line.extend_from_slice(data);
+                    self.total_size += data.len();
+                    return;
+                }
+            }
+        }
+
+        // Process remaining data
+        let remaining = &data[start..];
+        let ends_with_newline = remaining.last() == Some(&b'\n');
+
+        // Split by lines
+        let lines: Vec<&[u8]> = remaining.split(|&b| b == b'\n').collect();
+
+        for (i, line) in lines.iter().enumerate() {
+            if i == lines.len() - 1 && !ends_with_newline && !line.is_empty() {
+                // Last line without newline
+                self.last_line_incomplete = true;
+            }
+
+            if !line.is_empty() || i < lines.len() - 1 {
+                let mut line_data = line.to_vec();
+                if i < lines.len() - 1 || ends_with_newline {
+                    line_data.push(b'\n');
+                }
+
+                self.total_size += line_data.len();
+                self.lines.push_back(line_data);
+            }
+        }
+
+        // Trim old data if buffer is too large
+        while self.total_size > MAX_OUTPUT_BUFFER_SIZE || self.lines.len() > MAX_BUFFER_LINES {
+            if let Some(removed) = self.lines.pop_front() {
+                self.total_size -= removed.len();
+            }
+        }
+    }
+
+    fn get_recent(&self, max_bytes: usize) -> Vec<u8> {
+        let mut result = Vec::new();
+        let mut size = 0;
+
+        // Get recent lines up to max_bytes
+        for line in self.lines.iter().rev() {
+            if size + line.len() > max_bytes {
+                break;
+            }
+            size += line.len();
+            result.splice(0..0, line.iter().cloned());
+        }
+
+        result
+    }
+}
+
+pub struct TerminalSession {
+    pub created_at: Instant,
+    last_activity: Instant,
+    pty_pair: Option<portable_pty::PtyPair>,
+    child: Option<Box<dyn Child + std::marker::Send + Sync>>,
+    // Channel for sending input to the writer thread
+    input_tx: Option<SyncSender<Vec<u8>>>,
+    // Channel for receiving output from the reader thread
+    output_rx: Option<Receiver<Vec<u8>>>,
+    // Thread handles
+    reader_thread: Option<thread::JoinHandle<()>>,
+    writer_thread: Option<thread::JoinHandle<()>>,
+    output_buffer: OutputBuffer,
+    title: String,
+    pid: u32,
+    rows: u16,
+    cols: u16,
+    // Track if we've already sent the closed message
+    closed_message_sent: bool,
+}
+
+impl TerminalSession {
+    fn new(terminal_id: i32, rows: u16, cols: u16) -> Self {
+        Self {
+            created_at: Instant::now(),
+            last_activity: Instant::now(),
+            pty_pair: None,
+            child: None,
+            input_tx: None,
+            output_rx: None,
+            reader_thread: None,
+            writer_thread: None,
+            output_buffer: OutputBuffer::new(),
+            title: format!("Terminal {}", terminal_id),
+            pid: 0,
+            rows,
+            cols,
+            closed_message_sent: false,
+        }
+    }
+
+    fn update_activity(&mut self) {
+        self.last_activity = Instant::now();
+    }
+}
+
+impl Drop for TerminalSession {
+    fn drop(&mut self) {
+        // Drop the input channel to signal writer thread to exit
+        drop(self.input_tx.take());
+
+        // Wait for threads to finish
+        if let Some(writer_thread) = self.writer_thread.take() {
+            let _ = writer_thread.join();
+        }
+        if let Some(reader_thread) = self.reader_thread.take() {
+            let _ = reader_thread.join();
+        }
+
+        // Ensure child process is properly handled when session is dropped
+        if let Some(mut child) = self.child.take() {
+            let _ = child.kill();
+            add_to_reaper(child);
+        }
+    }
+}
+
+/// Persistent terminal service that can survive connection drops
+pub struct PersistentTerminalService {
+    service_id: String,
+    sessions: HashMap<i32, Arc<Mutex<TerminalSession>>>,
+    pub created_at: Instant,
+    last_activity: Instant,
+    pub is_persistent: bool,
+}
+
+impl PersistentTerminalService {
+    pub fn new(service_id: String, is_persistent: bool) -> Self {
+        Self {
+            service_id,
+            sessions: HashMap::new(),
+            created_at: Instant::now(),
+            last_activity: Instant::now(),
+            is_persistent,
+        }
+    }
+
+    fn update_activity(&mut self) {
+        self.last_activity = Instant::now();
+    }
+
+    /// Get list of terminal metadata
+    pub fn list_terminals(&self) -> Vec<(i32, String, u32, Instant)> {
+        self.sessions
+            .iter()
+            .map(|(id, session)| {
+                let s = session.lock().unwrap();
+                (*id, s.title.clone(), s.pid, s.created_at)
+            })
+            .collect()
+    }
+
+    /// Get buffered output for a terminal
+    pub fn get_terminal_buffer(&self, terminal_id: i32, max_bytes: usize) -> Option<Vec<u8>> {
+        self.sessions.get(&terminal_id).map(|session| {
+            let session = session.lock().unwrap();
+            session.output_buffer.get_recent(max_bytes)
+        })
+    }
+
+    /// Get terminal info for recovery
+    pub fn get_terminal_info(&self, terminal_id: i32) -> Option<(u16, u16, Vec<u8>)> {
+        self.sessions.get(&terminal_id).map(|session| {
+            let session = session.lock().unwrap();
+            (
+                session.rows,
+                session.cols,
+                session.output_buffer.get_recent(4096),
+            )
+        })
+    }
+
+    /// Check if service has active terminals
+    pub fn has_active_terminals(&self) -> bool {
+        !self.sessions.is_empty()
+    }
+}
+
+pub struct TerminalServiceProxy {
+    service_id: String,
+    is_persistent: bool,
+}
+
+pub fn set_persistent(service_id: &str, is_persistent: bool) -> Result<()> {
+    if let Some(service) = get_service(service_id) {
+        service.lock().unwrap().is_persistent = is_persistent;
+        Ok(())
+    } else {
+        Err(anyhow!("Service {} not found", service_id))
+    }
+}
+
+impl TerminalServiceProxy {
+    pub fn new(service_id: String, is_persistent: Option<bool>) -> Self {
+        // Get persistence from the service if it exists
+        let is_persistent =
+            is_persistent.unwrap_or(if let Some(service) = get_service(&service_id) {
+                service.lock().unwrap().is_persistent
+            } else {
+                false
+            });
+        TerminalServiceProxy {
+            service_id,
+            is_persistent,
+        }
+    }
+
+    pub fn get_service_id(&self) -> &str {
+        &self.service_id
+    }
+
+    pub fn handle_action(&mut self, action: &TerminalAction) -> Result<Option<TerminalResponse>> {
+        let service = match get_service(&self.service_id) {
+            Some(s) => s,
+            None => {
+                let mut response = TerminalResponse::new();
+                let mut error = TerminalError::new();
+                error.message = format!("Terminal service {} not found", self.service_id);
+                response.set_error(error);
+                return Ok(Some(response));
+            }
+        };
+        service.lock().unwrap().update_activity();
+        match &action.union {
+            Some(terminal_action::Union::Open(open)) => {
+                self.handle_open(&mut service.lock().unwrap(), open)
+            }
+            Some(terminal_action::Union::Resize(resize)) => {
+                let session = service
+                    .lock()
+                    .unwrap()
+                    .sessions
+                    .get(&resize.terminal_id)
+                    .cloned();
+                self.handle_resize(session, resize)
+            }
+            Some(terminal_action::Union::Data(data)) => {
+                let session = service
+                    .lock()
+                    .unwrap()
+                    .sessions
+                    .get(&data.terminal_id)
+                    .cloned();
+                self.handle_data(session, data)
+            }
+            Some(terminal_action::Union::Close(close)) => {
+                self.handle_close(&mut service.lock().unwrap(), close)
+            }
+            _ => Ok(None),
+        }
+    }
+
+    fn handle_open(
+        &self,
+        service: &mut PersistentTerminalService,
+        open: &OpenTerminal,
+    ) -> Result<Option<TerminalResponse>> {
+        let mut response = TerminalResponse::new();
+
+        // Check if terminal already exists
+        if let Some(session_arc) = service.sessions.get(&open.terminal_id) {
+            // Reconnect to existing terminal
+            let session = session_arc.lock().unwrap();
+            let mut opened = TerminalOpened::new();
+            opened.terminal_id = open.terminal_id;
+            opened.success = true;
+            opened.message = "Reconnected to existing terminal".to_string();
+            opened.pid = session.pid;
+            // Return service_id for persistent sessions
+            if self.is_persistent {
+                opened.service_id = self.service_id.clone();
+            }
+            response.set_opened(opened);
+
+            // Send buffered output
+            let buffer = session.output_buffer.get_recent(4096);
+            if !buffer.is_empty() {
+                // We'll need to send this separately or extend the protocol
+                // For now, just acknowledge the reconnection
+            }
+
+            return Ok(Some(response));
+        }
+
+        // Create new terminal session
+        log::info!(
+            "Creating new terminal {} for service: {}",
+            open.terminal_id,
+            service.service_id
+        );
+        let mut session =
+            TerminalSession::new(open.terminal_id, open.rows as u16, open.cols as u16);
+
+        let pty_size = PtySize {
+            rows: open.rows as u16,
+            cols: open.cols as u16,
+            pixel_width: 0,
+            pixel_height: 0,
+        };
+
+        log::debug!("Opening PTY with size: {}x{}", open.rows, open.cols);
+        let pty_system = portable_pty::native_pty_system();
+        let pty_pair = pty_system.openpty(pty_size).context("Failed to open PTY")?;
+
+        // Use default shell for the platform
+        let shell = get_default_shell();
+        log::debug!("Using shell: {}", shell);
+        let cmd = CommandBuilder::new(&shell);
+
+        log::debug!("Spawning shell process...");
+        let child = pty_pair
+            .slave
+            .spawn_command(cmd)
+            .context("Failed to spawn command")?;
+
+        let writer = pty_pair
+            .master
+            .take_writer()
+            .context("Failed to get writer")?;
+
+        let reader = pty_pair
+            .master
+            .try_clone_reader()
+            .context("Failed to get reader")?;
+
+        session.pid = child.process_id().unwrap_or(0) as u32;
+
+        // Create channels for input/output
+        let (input_tx, input_rx) = mpsc::sync_channel::<Vec<u8>>(CHANNEL_BUFFER_SIZE);
+        let (output_tx, output_rx) = mpsc::sync_channel::<Vec<u8>>(CHANNEL_BUFFER_SIZE);
+
+        // Spawn writer thread
+        let terminal_id = open.terminal_id;
+        let writer_thread = thread::spawn(move || {
+            let mut writer = writer;
+            while let Ok(data) = input_rx.recv() {
+                if let Err(e) = writer.write_all(&data) {
+                    log::error!("Terminal {} write error: {}", terminal_id, e);
+                    break;
+                }
+                if let Err(e) = writer.flush() {
+                    log::error!("Terminal {} flush error: {}", terminal_id, e);
+                }
+            }
+            log::debug!("Terminal {} writer thread exiting", terminal_id);
+        });
+
+        // Spawn reader thread
+        let terminal_id = open.terminal_id;
+        let reader_thread = thread::spawn(move || {
+            let mut reader = reader;
+            let mut buf = vec![0u8; 4096];
+            loop {
+                match reader.read(&mut buf) {
+                    Ok(0) => {
+                        // EOF
+                        break;
+                    }
+                    Ok(n) => {
+                        let data = buf[..n].to_vec();
+                        // Try to send, if channel is full, drop the data
+                        match output_tx.try_send(data) {
+                            Ok(_) => {}
+                            Err(mpsc::TrySendError::Full(_)) => {
+                                log::debug!(
+                                    "Terminal {} output channel full, dropping data",
+                                    terminal_id
+                                );
+                            }
+                            Err(mpsc::TrySendError::Disconnected(_)) => {
+                                log::debug!("Terminal {} output channel disconnected", terminal_id);
+                                break;
+                            }
+                        }
+                    }
+                    Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                        // For non-blocking I/O, sleep briefly
+                        thread::sleep(Duration::from_millis(10));
+                    }
+                    Err(e) => {
+                        log::error!("Terminal {} read error: {}", terminal_id, e);
+                        break;
+                    }
+                }
+            }
+            log::debug!("Terminal {} reader thread exiting", terminal_id);
+        });
+
+        session.pty_pair = Some(pty_pair);
+        session.child = Some(child);
+        session.input_tx = Some(input_tx);
+        session.output_rx = Some(output_rx);
+        session.reader_thread = Some(reader_thread);
+        session.writer_thread = Some(writer_thread);
+
+        let mut opened = TerminalOpened::new();
+        opened.terminal_id = open.terminal_id;
+        opened.success = true;
+        opened.message = "Terminal opened".to_string();
+        opened.pid = session.pid;
+        // Return service_id for persistent sessions
+        if self.is_persistent {
+            opened.service_id = service.service_id.clone();
+        }
+        response.set_opened(opened);
+
+        log::info!(
+            "Terminal {} opened successfully with PID {}",
+            open.terminal_id,
+            session.pid
+        );
+
+        // Store the session
+        service
+            .sessions
+            .insert(open.terminal_id, Arc::new(Mutex::new(session)));
+
+        Ok(Some(response))
+    }
+
+    fn handle_resize(
+        &self,
+        session: Option<Arc<Mutex<TerminalSession>>>,
+        resize: &ResizeTerminal,
+    ) -> Result<Option<TerminalResponse>> {
+        if let Some(session_arc) = session {
+            let mut session = session_arc.lock().unwrap();
+            session.update_activity();
+            session.rows = resize.rows as u16;
+            session.cols = resize.cols as u16;
+
+            if let Some(pty_pair) = &session.pty_pair {
+                pty_pair.master.resize(PtySize {
+                    rows: resize.rows as u16,
+                    cols: resize.cols as u16,
+                    pixel_width: 0,
+                    pixel_height: 0,
+                })?;
+            }
+        }
+        Ok(None)
+    }
+
+    fn handle_data(
+        &self,
+        session: Option<Arc<Mutex<TerminalSession>>>,
+        data: &TerminalData,
+    ) -> Result<Option<TerminalResponse>> {
+        if let Some(session_arc) = session {
+            let mut session = session_arc.lock().unwrap();
+            session.update_activity();
+            if let Some(input_tx) = &session.input_tx {
+                // Send data to writer thread
+                if let Err(e) = input_tx.send(data.data.to_vec()) {
+                    log::error!(
+                        "Failed to send data to terminal {}: {}",
+                        data.terminal_id,
+                        e
+                    );
+                }
+            }
+        }
+
+        Ok(None)
+    }
+
+    fn handle_close(
+        &self,
+        service: &mut PersistentTerminalService,
+        close: &CloseTerminal,
+    ) -> Result<Option<TerminalResponse>> {
+        let mut response = TerminalResponse::new();
+
+        // Always close and remove the terminal
+        if let Some(session_arc) = service.sessions.remove(&close.terminal_id) {
+            let mut session = session_arc.lock().unwrap();
+            let exit_code = if let Some(mut child) = session.child.take() {
+                child.kill()?;
+                add_to_reaper(child);
+                -1 // -1 indicates forced termination
+            } else {
+                0
+            };
+
+            let mut closed = TerminalClosed::new();
+            closed.terminal_id = close.terminal_id;
+            closed.exit_code = exit_code;
+            response.set_closed(closed);
+            Ok(Some(response))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub fn read_outputs(&self) -> Vec<TerminalResponse> {
+        let service = match get_service(&self.service_id) {
+            Some(s) => s,
+            None => {
+                return vec![];
+            }
+        };
+
+        // Get session references with minimal service lock time
+        let sessions: Vec<(i32, Arc<Mutex<TerminalSession>>)> = {
+            let service = service.lock().unwrap();
+            service
+                .sessions
+                .iter()
+                .map(|(id, session)| (*id, session.clone()))
+                .collect()
+        };
+
+        let mut responses = Vec::new();
+        let mut closed_terminals = Vec::new();
+
+        // Process each session with its own lock
+        for (terminal_id, session_arc) in sessions {
+            if let Ok(mut session) = session_arc.try_lock() {
+                // Check if reader thread is still alive and we haven't sent closed message yet
+                let mut should_send_closed = false;
+                if !session.closed_message_sent {
+                    if let Some(thread) = &session.reader_thread {
+                        if thread.is_finished() {
+                            should_send_closed = true;
+                            session.closed_message_sent = true;
+                        }
+                    }
+                }
+
+                // Read from output channel
+                let mut has_activity = false;
+                let mut received_data = Vec::new();
+                if let Some(output_rx) = &session.output_rx {
+                    // Try to read all available data
+                    while let Ok(data) = output_rx.try_recv() {
+                        has_activity = true;
+                        received_data.push(data);
+                    }
+                }
+
+                // Update buffer after reading
+                for data in &received_data {
+                    session.output_buffer.append(data);
+                }
+
+                // Process received data for responses
+                for data in received_data {
+                    let mut response = TerminalResponse::new();
+                    let mut terminal_data = TerminalData::new();
+                    terminal_data.terminal_id = terminal_id;
+
+                    // Compress data if it exceeds threshold
+                    if data.len() > COMPRESS_THRESHOLD {
+                        let compressed = compress::compress(&data);
+                        if compressed.len() < data.len() {
+                            terminal_data.data = bytes::Bytes::from(compressed);
+                            terminal_data.compressed = true;
+                        } else {
+                            // Compression didn't help, send uncompressed
+                            terminal_data.data = bytes::Bytes::from(data);
+                        }
+                    } else {
+                        terminal_data.data = bytes::Bytes::from(data);
+                    }
+
+                    response.set_data(terminal_data);
+                    responses.push(response);
+                }
+
+                if has_activity {
+                    session.update_activity();
+                }
+
+                if should_send_closed {
+                    closed_terminals.push(terminal_id);
+                }
+            }
+        }
+
+        // Clean up closed terminals (requires service lock briefly)
+        if !closed_terminals.is_empty() {
+            let mut sessions = service.lock().unwrap().sessions.clone();
+            for terminal_id in closed_terminals {
+                let mut exit_code = 0;
+
+                if !self.is_persistent {
+                    if let Some(session_arc) = sessions.remove(&terminal_id) {
+                        service.lock().unwrap().sessions.remove(&terminal_id);
+                        let mut session = session_arc.lock().unwrap();
+                        // Take the child and add to zombie reaper
+                        if let Some(mut child) = session.child.take() {
+                            // Try to get exit code if available
+                            if let Ok(Some(status)) = child.try_wait() {
+                                exit_code = status.exit_code() as i32;
+                            }
+                            add_to_reaper(child);
+                        }
+                    }
+                } else {
+                    // For persistent sessions, just clear the child reference
+                    if let Some(session_arc) = sessions.get(&terminal_id) {
+                        let mut session = session_arc.lock().unwrap();
+                        if let Some(mut child) = session.child.take() {
+                            // Try to get exit code if available
+                            if let Ok(Some(status)) = child.try_wait() {
+                                exit_code = status.exit_code() as i32;
+                            }
+                            add_to_reaper(child);
+                        }
+                    }
+                }
+
+                let mut response = TerminalResponse::new();
+                let mut closed = TerminalClosed::new();
+                closed.terminal_id = terminal_id;
+                closed.exit_code = exit_code;
+                response.set_closed(closed);
+                responses.push(response);
+            }
+        }
+
+        responses
+    }
+
+    /// Cleanup when connection drops
+    pub fn on_disconnect(&self) {
+        if !self.is_persistent {
+            // Remove non-persistent service
+            remove_service(&self.service_id);
+        }
+    }
+}