fix: terminal, restore, multi-sessions, msgs (#12364)

Signed-off-by: fufesou <linlong1266@gmail.com>

src/server/terminal_service.rs

@@ -131,7 +131,7 @@ fn get_or_create_service(
     // Ensure cleanup task is running
     ensure_cleanup_task();
 
-    service.lock().unwrap().needs_session_sync = true;
+    service.lock().unwrap().reset_status();
 
     Ok(service)
 }
@@ -447,6 +447,7 @@ pub struct TerminalSession {
     cols: u16,
     // Track if we've already sent the closed message
     closed_message_sent: bool,
+    is_opened: bool,
 }
 
 impl TerminalSession {
@@ -467,6 +468,7 @@ impl TerminalSession {
             rows,
             cols,
             closed_message_sent: false,
+            is_opened: false,
         }
     }
 
@@ -477,6 +479,7 @@ impl TerminalSession {
     // This helper function is to ensure that the threads are joined before the child process is dropped.
     // Though this is not strictly necessary on macOS.
     fn stop(&mut self) {
+        self.is_opened = false;
         self.exiting.store(true, Ordering::SeqCst);
 
         // Drop the input channel to signal writer thread to exit
@@ -596,6 +599,14 @@ impl PersistentTerminalService {
     pub fn has_active_terminals(&self) -> bool {
         !self.sessions.is_empty()
     }
+
+    fn reset_status(&mut self) {
+        self.needs_session_sync = true;
+        for session in self.sessions.values() {
+            let mut session = session.lock().unwrap();
+            session.is_opened = false;
+        }
+    }
 }
 
 pub struct TerminalServiceProxy {
@@ -690,7 +701,8 @@ impl TerminalServiceProxy {
         // 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 session = session_arc.lock().unwrap();
+            session.is_opened = true;
             let mut opened = TerminalOpened::new();
             opened.terminal_id = open.terminal_id;
             opened.success = true;
@@ -860,6 +872,7 @@ impl TerminalServiceProxy {
         session.output_rx = Some(output_rx);
         session.reader_thread = Some(reader_thread);
         session.writer_thread = Some(writer_thread);
+        session.is_opened = true;
 
         let mut opened = TerminalOpened::new();
         opened.terminal_id = open.terminal_id;
@@ -997,6 +1010,17 @@ impl TerminalServiceProxy {
                         }
                     }
                 }
+                // It's Ok to put the closed message here.
+                // Because the `reader_thread` is joined in `stop()`,
+                // and `stop()` is called before the session is dropped.
+                if should_send_closed {
+                    closed_terminals.push(terminal_id);
+                }
+
+                if !session.is_opened {
+                    // Skip the session if it is not opened.
+                    continue;
+                }
 
                 // Read from output channel
                 let mut has_activity = false;
@@ -1041,10 +1065,6 @@ impl TerminalServiceProxy {
                 if has_activity {
                     session.update_activity();
                 }
-
-                if should_send_closed {
-                    closed_terminals.push(terminal_id);
-                }
             }
         }