feat: add terminal session observe stream

3 files changed, 310 insertions, 8 deletions

diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index 0dea1179..b69490c8 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -312,6 +312,8 @@ pub(crate) enum ServerEvent {
         terminal_id: String,
         takeover: bool,
     },
+    /// A client requested read-only observation of one terminal.
+    ClientObserveTerminal { client_id: u64, target: String },
     /// A direct terminal attach client requested scrollback movement.
     ClientAttachScroll {
         client_id: u64,
@@ -668,6 +670,9 @@ fn client_read_loop(
                     ServerEvent::ClientInputEvents { client_id, events }
                 }
             }
+            ClientMessage::ObserveTerminal { target } => {
+                ServerEvent::ClientObserveTerminal { client_id, target }
+            }
             ClientMessage::ClipboardImage { extension, data } => {
                 if data.len() > MAX_CLIPBOARD_IMAGE_PAYLOAD {
                     warn!(
diff --git a/src/server/clients.rs b/src/server/clients.rs
index eb823cfc..705152eb 100644
--- a/src/server/clients.rs
+++ b/src/server/clients.rs
@@ -9,6 +9,7 @@ use crate::server::render_stream::ClientRenderState;
 pub(crate) enum ClientConnectionMode {
     App,
     TerminalAttach { terminal_id: String },
+    TerminalObserve { terminal_id: String },
 }
 
 pub(crate) type RenderTarget = (
@@ -220,7 +221,7 @@ pub(crate) fn latest_app_client(clients: &HashMap<u64, ClientConnection>) -> Opt
         .map(|(&client_id, _)| client_id)
 }
 
-pub(crate) fn terminal_attach_client_ids(
+pub(crate) fn terminal_stream_client_ids(
     clients: &HashMap<u64, ClientConnection>,
     terminal_id: &str,
 ) -> Vec<u64> {
@@ -229,6 +230,9 @@ pub(crate) fn terminal_attach_client_ids(
         .filter_map(|(&client_id, client)| match &client.mode {
             ClientConnectionMode::TerminalAttach {
                 terminal_id: attached,
+            }
+            | ClientConnectionMode::TerminalObserve {
+                terminal_id: attached,
             } if attached == terminal_id => Some(client_id),
             _ => None,
         })
@@ -244,7 +248,11 @@ pub(crate) fn render_targets(
         .filter(|(_, client)| {
             client.writer.is_some()
                 && (client.is_full_app_client()
-                    || matches!(client.mode, ClientConnectionMode::TerminalAttach { .. }))
+                    || matches!(
+                        client.mode,
+                        ClientConnectionMode::TerminalAttach { .. }
+                            | ClientConnectionMode::TerminalObserve { .. }
+                    ))
         })
         .map(|(&client_id, client)| {
             (
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 469320f9..6b68a7b6 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -52,7 +52,7 @@ use crate::server::client_accept::{
 };
 use crate::server::client_transport::ServerEvent;
 use crate::server::clients::{
-    events_include_interaction, latest_app_client, render_targets, terminal_attach_client_ids,
+    events_include_interaction, latest_app_client, render_targets, terminal_stream_client_ids,
     ClientConnection, ClientConnectionMode,
 };
 use crate::server::keybindings::{app_keybindings, apply_keybindings};
@@ -1180,7 +1180,10 @@ impl HeadlessServer {
         }
         matches!(
             self.clients.get(&client_id).map(|client| &client.mode),
-            Some(ClientConnectionMode::TerminalAttach { .. })
+            Some(
+                ClientConnectionMode::TerminalAttach { .. }
+                    | ClientConnectionMode::TerminalObserve { .. }
+            )
         ) && self.foreground_client_id.is_some()
     }
 
@@ -1312,6 +1315,16 @@ impl HeadlessServer {
         self.app.terminal_runtimes.get(&terminal_id)
     }
 
+    fn resolve_terminal_target_id_string(&self, target: &str) -> Option<String> {
+        if self.terminal_id_by_string(target).is_some() {
+            return Some(target.to_owned());
+        }
+        self.app
+            .resolve_terminal_target(target)
+            .ok()
+            .map(|resolved| resolved.terminal_id)
+    }
+
     fn write_client_clipboard_image(
         &mut self,
         client_id: u64,
@@ -1355,6 +1368,54 @@ impl HeadlessServer {
         true
     }
 
+    fn observe_terminal_client(&mut self, client_id: u64, target: String) -> bool {
+        if !self.client_is_pending_terminal_mode(client_id) {
+            self.send_to_client(
+                client_id,
+                ServerMessage::ServerShutdown {
+                    reason: Some(
+                        "terminal session observe failed: connection is not pending terminal session"
+                            .to_owned(),
+                    ),
+                },
+            );
+            self.remove_client_and_resize_if_needed(client_id);
+            return false;
+        }
+
+        let Some(terminal_id) = self.resolve_terminal_target_id_string(&target) else {
+            self.send_to_client(
+                client_id,
+                ServerMessage::ServerShutdown {
+                    reason: Some(format!(
+                        "terminal session observe failed: terminal target {target} not found"
+                    )),
+                },
+            );
+            self.remove_client_and_resize_if_needed(client_id);
+            return false;
+        };
+
+        let stamp = self.allocate_activity_stamp();
+        let Some(client) = self.clients.get_mut(&client_id) else {
+            return false;
+        };
+        let (cols, rows) = client.terminal_size;
+        client.mode = ClientConnectionMode::TerminalObserve {
+            terminal_id: terminal_id.clone(),
+        };
+        client.pending_terminal_attach = false;
+        client.render_state.reset_baseline();
+        client.last_activity = stamp;
+        let was_foreground = self.foreground_client_id == Some(client_id);
+        if was_foreground {
+            self.promote_latest_remaining_client();
+        }
+
+        info!(client_id, cols, rows, terminal_id = %terminal_id, "terminal observe client connected");
+        true
+    }
+
     fn handle_terminal_attach_scroll(
         &mut self,
         client_id: u64,
@@ -1915,7 +1976,7 @@ impl HeadlessServer {
 
                 if self.app.find_pane(pane_id_val).is_none() {
                     if let Some(terminal_id) = terminal_id {
-                        self.shutdown_terminal_attach_clients(
+                        self.shutdown_terminal_stream_clients(
                             &terminal_id,
                             format!("terminal {terminal_id} exited"),
                         );
@@ -2071,8 +2132,8 @@ impl HeadlessServer {
         }
     }
 
-    fn shutdown_terminal_attach_clients(&mut self, terminal_id: &str, reason: String) {
-        let client_ids = terminal_attach_client_ids(&self.clients, terminal_id);
+    fn shutdown_terminal_stream_clients(&mut self, terminal_id: &str, reason: String) {
+        let client_ids = terminal_stream_client_ids(&self.clients, terminal_id);
 
         for client_id in client_ids {
             self.send_to_client(
@@ -2114,6 +2175,20 @@ impl HeadlessServer {
         terminal_id: String,
         takeover: bool,
     ) -> bool {
+        if !self.client_is_pending_terminal_mode(client_id) {
+            self.send_to_client(
+                client_id,
+                ServerMessage::ServerShutdown {
+                    reason: Some(
+                        "terminal attach failed: connection is not pending terminal attach"
+                            .to_owned(),
+                    ),
+                },
+            );
+            self.remove_client_and_resize_if_needed(client_id);
+            return false;
+        }
+
         let Some(real_terminal_id) = self.terminal_id_by_string(&terminal_id) else {
             self.send_to_client(
                 client_id,
@@ -2183,6 +2258,12 @@ impl HeadlessServer {
         true
     }
 
+    fn client_is_pending_terminal_mode(&self, client_id: u64) -> bool {
+        self.clients.get(&client_id).is_some_and(|client| {
+            client.pending_terminal_attach && matches!(client.mode, ClientConnectionMode::App)
+        })
+    }
+
     /// Handles a server event. Returns true if the event requires a re-render.
     fn handle_client_input_events(
         &mut self,
@@ -2321,6 +2402,9 @@ impl HeadlessServer {
                 terminal_id,
                 takeover,
             } => self.attach_terminal_client(client_id, terminal_id, takeover),
+            ServerEvent::ClientObserveTerminal { client_id, target } => {
+                self.observe_terminal_client(client_id, target)
+            }
             ServerEvent::ClientAttachScroll {
                 client_id,
                 source,
@@ -2354,6 +2438,12 @@ impl HeadlessServer {
                     }
                     return true;
                 }
+                if matches!(
+                    self.clients.get(&client_id).map(|client| &client.mode),
+                    Some(ClientConnectionMode::TerminalObserve { .. })
+                ) {
+                    return false;
+                }
                 let events = if let Some(client) = self.clients.get_mut(&client_id) {
                     let mut events = client.raw_input.push(&data);
                     // The thin client only forwards a bare ESC after its local input timeout.
@@ -2380,6 +2470,12 @@ impl HeadlessServer {
                     len = events.len(),
                     "client input events received"
                 );
+                if matches!(
+                    self.clients.get(&client_id).map(|client| &client.mode),
+                    Some(ClientConnectionMode::TerminalObserve { .. })
+                ) {
+                    return false;
+                }
                 let events = events
                     .iter()
                     .map(crate::protocol::ClientInputEvent::to_raw_input_event)
@@ -2397,6 +2493,12 @@ impl HeadlessServer {
                     extension = %extension,
                     "client clipboard image received"
                 );
+                if matches!(
+                    self.clients.get(&client_id).map(|client| &client.mode),
+                    Some(ClientConnectionMode::TerminalObserve { .. })
+                ) {
+                    return false;
+                }
                 match self.write_client_clipboard_image(client_id, &extension, &data) {
                     Ok(path) => self.paste_client_clipboard_image_path(client_id, path),
                     Err(err) => {
@@ -2440,6 +2542,22 @@ impl HeadlessServer {
                     }
                     return true;
                 }
+                if let Some(ClientConnection {
+                    mode: ClientConnectionMode::TerminalObserve { .. },
+                    terminal_size,
+                    cell_size,
+                    render_state,
+                    ..
+                }) = self.clients.get_mut(&client_id)
+                {
+                    *terminal_size = (cols, rows);
+                    *cell_size = crate::kitty_graphics::HostCellSize {
+                        width_px: cell_width_px,
+                        height_px: cell_height_px,
+                    };
+                    render_state.reset_baseline();
+                    return true;
+                }
                 if let Some(client) = self.clients.get_mut(&client_id) {
                     client.terminal_size = (cols, rows);
                     client.cell_size = crate::kitty_graphics::HostCellSize {
@@ -3120,7 +3238,8 @@ impl HeadlessServer {
                     crate::render_prof::duration_since("full_render.frame_build", frame_started);
                     frame
                 }
-                ClientConnectionMode::TerminalAttach { terminal_id } => {
+                ClientConnectionMode::TerminalAttach { terminal_id }
+                | ClientConnectionMode::TerminalObserve { terminal_id } => {
                     let Some(runtime) = self.runtime_for_terminal_id_string(&terminal_id) else {
                         self.send_to_client(
                             client_id,
@@ -4388,6 +4507,176 @@ next_tab = ""
         );
     }
 
+    fn with_terminal_session_test_server(
+        test: impl FnOnce(&mut HeadlessServer, crate::terminal::TerminalId, String, String),
+    ) {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime");
+        let _runtime_guard = rt.enter();
+        let mut server = test_headless_server();
+        let workspace = crate::workspace::Workspace::test_new("test");
+        let pane_id = workspace.tabs[0].root_pane;
+        let terminal_id = workspace.terminal_id(pane_id).expect("terminal id").clone();
+        let terminal_id_string = terminal_id.to_string();
+        let public_pane_id = format!("{}:p1", workspace.id);
+        server.app.state.workspaces = vec![workspace];
+        server.app.state.ensure_test_terminals();
+        server.app.terminal_runtimes.insert(
+            terminal_id.clone(),
+            crate::terminal::TerminalRuntime::test_with_screen_bytes(80, 24, b""),
+        );
+
+        test(&mut server, terminal_id, terminal_id_string, public_pane_id);
+
+        drop(server);
+        drop(_runtime_guard);
+        rt.shutdown_timeout(Duration::from_millis(100));
+    }
+
+    fn connect_pending_terminal_client(server: &mut HeadlessServer, client_id: u64) {
+        let (writer, _control_rx, _render_rx) = test_client_writer();
+        assert!(server.handle_server_event(ServerEvent::ClientConnected {
+            client_id,
+            cols: 100,
+            rows: 30,
+            cell_width_px: 0,
+            cell_height_px: 0,
+            render_encoding: RenderEncoding::TerminalAnsi,
+            keybindings: None,
+            direct_attach_requested: true,
+            writer,
+        }));
+    }
+
+    #[test]
+    fn terminal_observe_allows_multiple_clients_without_attach_ownership() {
+        with_terminal_session_test_server(|server, terminal_id, terminal_id_string, _| {
+            let initial_size = server
+                .app
+                .terminal_runtimes
+                .get(&terminal_id)
+                .expect("runtime")
+                .current_size();
+
+            for client_id in [7, 8] {
+                connect_pending_terminal_client(server, client_id);
+                assert!(
+                    server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                        client_id,
+                        target: terminal_id_string.clone(),
+                    })
+                );
+            }
+
+            assert!(server.terminal_attach_owners.is_empty());
+            assert!(!server
+                .app
+                .state
+                .direct_attach_resize_locks
+                .contains(&terminal_id));
+            assert_eq!(
+                server
+                    .app
+                    .terminal_runtimes
+                    .get(&terminal_id)
+                    .expect("runtime")
+                    .current_size(),
+                initial_size
+            );
+            assert_eq!(
+                terminal_stream_client_ids(&server.clients, &terminal_id_string).len(),
+                2
+            );
+        });
+    }
+
+    #[test]
+    fn terminal_observe_resolves_public_pane_id() {
+        with_terminal_session_test_server(|server, terminal_id, _, public_pane_id| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                    client_id: 7,
+                    target: public_pane_id,
+                })
+            );
+
+            assert!(matches!(
+                server.clients.get(&7).map(|client| &client.mode),
+                Some(ClientConnectionMode::TerminalObserve { terminal_id: observed })
+                    if observed == &terminal_id.to_string()
+            ));
+        });
+    }
+
+    #[test]
+    fn terminal_observe_rejects_later_attach_upgrade() {
+        with_terminal_session_test_server(|server, terminal_id, terminal_id_string, _| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                })
+            );
+            assert!(
+                !server.handle_server_event(ServerEvent::ClientAttachTerminal {
+                    client_id: 7,
+                    terminal_id: terminal_id_string,
+                    takeover: true,
+                })
+            );
+
+            assert!(!server.clients.contains_key(&7));
+            assert!(server.terminal_attach_owners.is_empty());
+            assert!(!server
+                .app
+                .state
+                .direct_attach_resize_locks
+                .contains(&terminal_id));
+        });
+    }
+
+    #[test]
+    fn terminal_attach_rejects_later_observe_and_clears_ownership() {
+        with_terminal_session_test_server(|server, terminal_id, terminal_id_string, _| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientAttachTerminal {
+                    client_id: 7,
+                    terminal_id: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+            assert_eq!(
+                server.terminal_attach_owners.get(&terminal_id_string),
+                Some(&7)
+            );
+            assert!(server
+                .app
+                .state
+                .direct_attach_resize_locks
+                .contains(&terminal_id));
+
+            assert!(
+                !server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                })
+            );
+
+            assert!(!server.clients.contains_key(&7));
+            assert!(server.terminal_attach_owners.is_empty());
+            assert!(!server
+                .app
+                .state
+                .direct_attach_resize_locks
+                .contains(&terminal_id));
+        });
+    }
+
     fn app_client_marks_git_refresh_due_on_first_attach(render_encoding: RenderEncoding) {
         let mut server = test_headless_server();
         server