feat: add terminal session control stream

5 files changed, 580 insertions, 29 deletions

diff --git a/src/cli.rs b/src/cli.rs
index 151286d1..64c25596 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -22,6 +22,8 @@ mod worktree;
 
 const TERMINAL_SESSION_OBSERVE_USAGE: &str =
     "usage: herdr terminal session observe <target> [--cols N] [--rows N]";
+const TERMINAL_SESSION_CONTROL_USAGE: &str =
+    "usage: herdr terminal session control <target> [--takeover] [--cols N] [--rows N]";
 
 pub(crate) fn parse_env_assignment(raw: &str) -> Result<(String, String), String> {
     let Some((key, value)) = raw.split_once('=') else {
@@ -482,66 +484,125 @@ fn terminal_attach(args: &[String]) -> std::io::Result<i32> {
 
 fn terminal_session(args: &[String]) -> std::io::Result<i32> {
     match args.first().map(|arg| arg.as_str()) {
+        Some("control") => terminal_session_control(&args[1..]),
         Some("observe") => terminal_session_observe(&args[1..]),
         Some("help" | "--help" | "-h") => {
+            eprintln!("{TERMINAL_SESSION_CONTROL_USAGE}");
             eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
             Ok(0)
         }
         _ => {
+            eprintln!("{TERMINAL_SESSION_CONTROL_USAGE}");
             eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
             Ok(2)
         }
     }
 }
 
+fn terminal_session_control(args: &[String]) -> std::io::Result<i32> {
+    let options = match parse_terminal_session_options(
+        args,
+        TERMINAL_SESSION_CONTROL_USAGE,
+        "control",
+        true,
+    )? {
+        Ok(options) => options,
+        Err(code) => return Ok(code),
+    };
+
+    crate::client::run_terminal_session_control(
+        options.target,
+        options.takeover,
+        options.cols,
+        options.rows,
+    )?;
+    Ok(0)
+}
+
 fn terminal_session_observe(args: &[String]) -> std::io::Result<i32> {
+    let options = match parse_terminal_session_options(
+        args,
+        TERMINAL_SESSION_OBSERVE_USAGE,
+        "observe",
+        false,
+    )? {
+        Ok(options) => options,
+        Err(code) => return Ok(code),
+    };
+
+    crate::client::run_terminal_session_observe(options.target, options.cols, options.rows)?;
+    Ok(0)
+}
+
+struct TerminalSessionOptions {
+    target: String,
+    cols: u16,
+    rows: u16,
+    takeover: bool,
+}
+
+fn parse_terminal_session_options(
+    args: &[String],
+    usage: &str,
+    command: &str,
+    allow_takeover: bool,
+) -> std::io::Result<Result<TerminalSessionOptions, i32>> {
     if matches!(
         args.first().map(|arg| arg.as_str()),
         Some("help" | "--help" | "-h")
     ) {
-        eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-        return Ok(0);
+        eprintln!("{usage}");
+        return Ok(Err(0));
     }
     let Some(target) = args.first() else {
-        eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-        return Ok(2);
+        eprintln!("{usage}");
+        return Ok(Err(2));
     };
 
     let mut cols = 120;
     let mut rows = 40;
+    let mut takeover = false;
     let mut i = 1;
     while i < args.len() {
         match args[i].as_str() {
+            "--takeover" if allow_takeover => {
+                takeover = true;
+                i += 1;
+            }
             "--cols" => {
                 let Some(value) = args.get(i + 1) else {
-                    eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-                    return Ok(2);
+                    eprintln!("{usage}");
+                    return Ok(Err(2));
                 };
                 cols = parse_terminal_dimension(value, "--cols")?;
                 i += 2;
             }
             "--rows" => {
                 let Some(value) = args.get(i + 1) else {
-                    eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-                    return Ok(2);
+                    eprintln!("{usage}");
+                    return Ok(Err(2));
                 };
                 rows = parse_terminal_dimension(value, "--rows")?;
                 i += 2;
             }
             "help" | "--help" | "-h" => {
-                eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-                return Ok(0);
+                eprintln!("{usage}");
+                return Ok(Err(0));
             }
             other => {
-                eprintln!("unknown terminal session observe option: {other}");
-                eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
-                return Ok(2);
+                eprintln!("unknown terminal session {command} option: {other}");
+                eprintln!("{usage}");
+                return Ok(Err(2));
             }
         }
     }
 
-    crate::client::run_terminal_session_observe(target.clone(), cols, rows)?;
-    Ok(0)
+    Ok(Ok(TerminalSessionOptions {
+        target: target.clone(),
+        cols,
+        rows,
+        takeover,
+    }))
 }
 
 fn parse_terminal_dimension(raw: &str, flag: &str) -> std::io::Result<u16> {
@@ -1000,6 +1061,7 @@ fn print_config_help() {
 fn print_terminal_help() {
     eprintln!("herdr terminal commands:");
     eprintln!("  herdr terminal attach <terminal_id> [--takeover]");
+    eprintln!("  herdr terminal session control <target> [--takeover] [--cols N] [--rows N]");
     eprintln!("  herdr terminal session observe <target> [--cols N] [--rows N]");
     eprintln!("  herdr terminal title set <title>");
     eprintln!("  herdr terminal title clear");
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 98105781..59b7b114 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -15,7 +15,7 @@
 mod input;
 
 use std::collections::HashSet;
-use std::io::{self, Write as _};
+use std::io::{self, BufRead, Write as _};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
@@ -36,12 +36,13 @@ use tracing::{debug, info, warn};
 
 use crate::ipc::LocalStream;
 use crate::protocol::render_ansi;
+#[cfg(unix)]
+use crate::protocol::MAX_CLIPBOARD_IMAGE_PAYLOAD;
 use crate::protocol::{
-    self, ClientKeybindings, ClientLaunchMode, ClientMessage, NotifyKind, RenderEncoding,
-    ServerMessage, MAX_FRAME_SIZE, MAX_GRAPHICS_FRAME_SIZE, PROTOCOL_VERSION,
+    self, AttachScrollDirection, AttachScrollSource, ClientKeybindings, ClientLaunchMode,
+    ClientMessage, NotifyKind, RenderEncoding, ServerMessage, MAX_FRAME_SIZE,
+    MAX_GRAPHICS_FRAME_SIZE, PROTOCOL_VERSION,
 };
-#[cfg(unix)]
-use crate::protocol::{AttachScrollDirection, AttachScrollSource, MAX_CLIPBOARD_IMAGE_PAYLOAD};
 use crate::server::socket_paths::client_socket_path;
 
 static RECEIVED_KITTY_GRAPHICS_IDS: OnceLock<Mutex<HashSet<u32>>> = OnceLock::new();
@@ -817,11 +818,70 @@ pub fn run_terminal_attach(_terminal_id: String, _takeover: bool) -> io::Result<
 
 /// Runs a read-only terminal session observer and prints one JSON envelope per frame.
 pub fn run_terminal_session_observe(target: String, cols: u16, rows: u16) -> io::Result<()> {
+    let mut stream =
+        connect_terminal_session_stream(target.clone(), cols, rows, "observing terminal session")?;
+    write_to_server(&mut stream, &ClientMessage::ObserveTerminal { target })?;
+    write_terminal_session_output(stream)
+}
+
+/// Runs a writable terminal session controller.
+pub fn run_terminal_session_control(
+    target: String,
+    takeover: bool,
+    cols: u16,
+    rows: u16,
+) -> io::Result<()> {
+    let mut stream = connect_terminal_session_stream(
+        target.clone(),
+        cols,
+        rows,
+        "controlling terminal session",
+    )?;
+    write_to_server(
+        &mut stream,
+        &ClientMessage::ControlTerminal { target, takeover },
+    )?;
+
+    let mut write_stream = stream.try_clone()?;
+    let _input_thread = std::thread::spawn(move || {
+        let stdin = io::stdin();
+        for line in stdin.lock().lines() {
+            let Ok(line) = line else {
+                break;
+            };
+            if line.trim().is_empty() {
+                continue;
+            }
+            match terminal_control_command_from_json(&line) {
+                Ok(message) => {
+                    let release = matches!(message, ClientMessage::Detach);
+                    if write_to_server(&mut write_stream, &message).is_err() {
+                        return;
+                    }
+                    if release {
+                        return;
+                    }
+                }
+                Err(err) => eprintln!("herdr: terminal session control input ignored: {err}"),
+            }
+        }
+        let _ = write_to_server(&mut write_stream, &ClientMessage::Detach);
+    });
+
+    write_terminal_session_output(stream)
+}
+
+fn connect_terminal_session_stream(
+    target: String,
+    cols: u16,
+    rows: u16,
+    log_message: &'static str,
+) -> io::Result<LocalStream> {
     init_logging();
 
     let socket_path = client_socket_path();
     crate::logging::startup("client");
-    info!(path = %socket_path.display(), target = %target, cols, rows, "observing terminal session");
+    info!(path = %socket_path.display(), target = %target, cols, rows, "{log_message}");
 
     let mut stream = match crate::ipc::connect_local_stream(&socket_path) {
         Ok(stream) => stream,
@@ -853,9 +913,11 @@ pub fn run_terminal_session_observe(target: String, cols: u16, rows: u16) -> io:
         }
     }
 
-    write_to_server(&mut stream, &ClientMessage::ObserveTerminal { target })?;
     stream.set_nonblocking(false)?;
+    Ok(stream)
+}
 
+fn write_terminal_session_output(mut stream: LocalStream) -> io::Result<()> {
     let mut stdout = io::stdout().lock();
     loop {
         match protocol::read_message(&mut stream, MAX_GRAPHICS_FRAME_SIZE) {
@@ -892,6 +954,125 @@ pub fn run_terminal_session_observe(target: String, cols: u16, rows: u16) -> io:
     }
 }
 
+#[derive(serde::Deserialize)]
+#[serde(tag = "type")]
+enum TerminalControlCommand {
+    #[serde(rename = "terminal.input")]
+    Input {
+        text: Option<String>,
+        bytes: Option<String>,
+    },
+    #[serde(rename = "terminal.resize")]
+    Resize {
+        cols: u16,
+        rows: u16,
+        #[serde(default)]
+        cell_width_px: u32,
+        #[serde(default)]
+        cell_height_px: u32,
+    },
+    #[serde(rename = "terminal.scroll")]
+    Scroll {
+        direction: TerminalControlScrollDirection,
+        lines: u16,
+        #[serde(default)]
+        source: TerminalControlScrollSource,
+        #[serde(default)]
+        column: Option<u16>,
+        #[serde(default)]
+        row: Option<u16>,
+        #[serde(default)]
+        modifiers: u8,
+    },
+    #[serde(rename = "terminal.release")]
+    Release {},
+}
+
+#[derive(Clone, Copy, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum TerminalControlScrollDirection {
+    Up,
+    Down,
+}
+
+#[derive(Clone, Copy, Default, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum TerminalControlScrollSource {
+    #[default]
+    Wheel,
+    PageKey,
+}
+
+fn terminal_control_command_from_json(raw: &str) -> Result<ClientMessage, String> {
+    let command = serde_json::from_str::<TerminalControlCommand>(raw)
+        .map_err(|err| format!("invalid json command: {err}"))?;
+    match command {
+        TerminalControlCommand::Input { text, bytes } => {
+            let data = match (text, bytes) {
+                (Some(_), Some(_)) => {
+                    return Err("terminal.input accepts text or bytes, not both".into())
+                }
+                (Some(text), None) => text.into_bytes(),
+                (None, Some(bytes)) => base64::engine::general_purpose::STANDARD
+                    .decode(bytes)
+                    .map_err(|err| format!("invalid terminal.input bytes: {err}"))?,
+                (None, None) => Vec::new(),
+            };
+            Ok(ClientMessage::Input { data })
+        }
+        TerminalControlCommand::Resize {
+            cols,
+            rows,
+            cell_width_px,
+            cell_height_px,
+        } => {
+            if cols == 0 || rows == 0 {
+                return Err("terminal.resize cols and rows must be greater than 0".into());
+            }
+            Ok(ClientMessage::Resize {
+                cols,
+                rows,
+                cell_width_px,
+                cell_height_px,
+            })
+        }
+        TerminalControlCommand::Scroll {
+            direction,
+            lines,
+            source,
+            column,
+            row,
+            modifiers,
+        } => {
+            if lines == 0 {
+                return Err("terminal.scroll lines must be greater than 0".into());
+            }
+            let direction = match direction {
+                TerminalControlScrollDirection::Up => AttachScrollDirection::Up,
+                TerminalControlScrollDirection::Down => AttachScrollDirection::Down,
+            };
+            let source = match source {
+                TerminalControlScrollSource::Wheel => AttachScrollSource::Wheel,
+                TerminalControlScrollSource::PageKey => AttachScrollSource::PageKey {
+                    input: match direction {
+                        AttachScrollDirection::Up => b"\x1b[5~".to_vec(),
+                        AttachScrollDirection::Down => b"\x1b[6~".to_vec(),
+                    },
+                },
+            };
+            Ok(ClientMessage::AttachScroll {
+                source,
+                direction,
+                lines,
+                column,
+                row,
+                modifiers,
+            })
+        }
+        TerminalControlCommand::Release {} => Ok(ClientMessage::Detach),
+    }
+}
+
 fn run_client_with_mode(
     requested_encoding: RenderEncoding,
     attach_request: Option<(String, bool)>,
@@ -2334,6 +2515,69 @@ mod tests {
     }
 
     #[test]
+    fn terminal_control_input_command_accepts_text() {
+        let action =
+            terminal_control_command_from_json(r#"{"type":"terminal.input","text":"hello"}"#)
+                .unwrap();
+        let ClientMessage::Input { data } = action else {
+            panic!("expected input command");
+        };
+        assert_eq!(data, b"hello");
+    }
+
+    #[test]
+    fn terminal_control_input_command_accepts_base64_bytes() {
+        let action =
+            terminal_control_command_from_json(r#"{"type":"terminal.input","bytes":"G1tB"}"#)
+                .unwrap();
+        let ClientMessage::Input { data } = action else {
+            panic!("expected input command");
+        };
+        assert_eq!(data, b"\x1b[A");
+    }
+
+    #[test]
+    fn terminal_control_resize_command_maps_to_client_resize() {
+        let action = terminal_control_command_from_json(
+            r#"{"type":"terminal.resize","cols":100,"rows":30,"cell_width_px":8,"cell_height_px":16}"#,
+        )
+        .unwrap();
+        let ClientMessage::Resize {
+            cols,
+            rows,
+            cell_width_px,
+            cell_height_px,
+        } = action
+        else {
+            panic!("expected resize command");
+        };
+        assert_eq!(
+            (cols, rows, cell_width_px, cell_height_px),
+            (100, 30, 8, 16)
+        );
+    }
+
+    #[test]
+    fn terminal_control_scroll_command_maps_to_attach_scroll() {
+        let action = terminal_control_command_from_json(
+            r#"{"type":"terminal.scroll","direction":"up","lines":3}"#,
+        )
+        .unwrap();
+        let ClientMessage::AttachScroll {
+            source,
+            direction,
+            lines,
+            ..
+        } = action
+        else {
+            panic!("expected scroll command");
+        };
+        assert_eq!(source, AttachScrollSource::Wheel);
+        assert_eq!(direction, AttachScrollDirection::Up);
+        assert_eq!(lines, 3);
+    }
+
+    #[test]
     fn forward_clipboard_uses_local_clipboard_path() {
         unsafe {
             std::env::set_var("SSH_CONNECTION", "1 2 3 4");
diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs
index e6ba5746..5f44691b 100644
--- a/src/protocol/wire.rs
+++ b/src/protocol/wire.rs
@@ -387,6 +387,14 @@ pub enum ClientMessage {
         /// Pane, terminal, or agent target to observe.
         target: String,
     },
+
+    /// Switch this connection into writable terminal control mode.
+    ControlTerminal {
+        /// Pane, terminal, or agent target to control.
+        target: String,
+        /// Replace an existing writable controller for this terminal.
+        takeover: bool,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -1005,6 +1013,13 @@ mod tests {
             }),
             8
         );
+        assert_eq!(
+            tag(&ClientMessage::ControlTerminal {
+                target: "w1:p1".to_owned(),
+                takeover: false,
+            }),
+            9
+        );
     }
 
     #[test]
@@ -1144,6 +1159,18 @@ mod tests {
     }
 
     #[test]
+    fn client_control_terminal_roundtrip() {
+        let msg = ClientMessage::ControlTerminal {
+            target: "w1:p1".to_owned(),
+            takeover: true,
+        };
+        let encoded = bincode::serde::encode_to_vec(&msg, bincode::config::standard()).unwrap();
+        let (decoded, _): (ClientMessage, _) =
+            bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
+        assert_eq!(msg, decoded);
+    }
+
+    #[test]
     fn client_attach_scroll_roundtrip() {
         let msg = ClientMessage::AttachScroll {
             source: AttachScrollSource::Wheel,
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index b69490c8..c4e43e0d 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -314,6 +314,12 @@ pub(crate) enum ServerEvent {
     },
     /// A client requested read-only observation of one terminal.
     ClientObserveTerminal { client_id: u64, target: String },
+    /// A client requested writable control of one terminal.
+    ClientControlTerminal {
+        client_id: u64,
+        target: String,
+        takeover: bool,
+    },
     /// A direct terminal attach client requested scrollback movement.
     ClientAttachScroll {
         client_id: u64,
@@ -673,6 +679,13 @@ fn client_read_loop(
             ClientMessage::ObserveTerminal { target } => {
                 ServerEvent::ClientObserveTerminal { client_id, target }
             }
+            ClientMessage::ControlTerminal { target, takeover } => {
+                ServerEvent::ClientControlTerminal {
+                    client_id,
+                    target,
+                    takeover,
+                }
+            }
             ClientMessage::ClipboardImage { extension, data } => {
                 if data.len() > MAX_CLIPBOARD_IMAGE_PAYLOAD {
                     warn!(
diff --git a/src/server/headless.rs b/src/server/headless.rs
index 6b68a7b6..5798913d 100644
--- a/src/server/headless.rs
+++ b/src/server/headless.rs
@@ -1368,31 +1368,46 @@ impl HeadlessServer {
         true
     }
 
-    fn observe_terminal_client(&mut self, client_id: u64, target: String) -> bool {
+    fn resolve_terminal_session_target(
+        &mut self,
+        client_id: u64,
+        target: &str,
+        action: &str,
+    ) -> Option<String> {
         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(),
+                        format!(
+                            "terminal session {action} failed: connection is not pending terminal session"
+                        ),
                     ),
                 },
             );
             self.remove_client_and_resize_if_needed(client_id);
-            return false;
+            return None;
         }
 
-        let Some(terminal_id) = self.resolve_terminal_target_id_string(&target) else {
+        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"
+                        "terminal session {action} failed: terminal target {target} not found"
                     )),
                 },
             );
             self.remove_client_and_resize_if_needed(client_id);
+            return None;
+        };
+
+        Some(terminal_id)
+    }
+
+    fn observe_terminal_client(&mut self, client_id: u64, target: String) -> bool {
+        let Some(terminal_id) = self.resolve_terminal_session_target(client_id, &target, "observe")
+        else {
             return false;
         };
 
@@ -1416,6 +1431,15 @@ impl HeadlessServer {
         true
     }
 
+    fn control_terminal_client(&mut self, client_id: u64, target: String, takeover: bool) -> bool {
+        let Some(terminal_id) = self.resolve_terminal_session_target(client_id, &target, "control")
+        else {
+            return false;
+        };
+
+        self.attach_terminal_client(client_id, terminal_id, takeover)
+    }
+
     fn handle_terminal_attach_scroll(
         &mut self,
         client_id: u64,
@@ -2146,6 +2170,23 @@ impl HeadlessServer {
         }
     }
 
+    fn send_terminal_stream_detach_shutdown(&mut self, client_id: u64) {
+        if matches!(
+            self.clients.get(&client_id).map(|client| &client.mode),
+            Some(
+                ClientConnectionMode::TerminalAttach { .. }
+                    | ClientConnectionMode::TerminalObserve { .. }
+            )
+        ) {
+            self.send_to_client(
+                client_id,
+                ServerMessage::ServerShutdown {
+                    reason: Some("detached".to_owned()),
+                },
+            );
+        }
+    }
+
     #[cfg(unix)]
     fn disconnect_all_clients_for_handoff(&mut self) {
         let client_ids = self.clients.keys().copied().collect::<Vec<_>>();
@@ -2405,6 +2446,11 @@ impl HeadlessServer {
             ServerEvent::ClientObserveTerminal { client_id, target } => {
                 self.observe_terminal_client(client_id, target)
             }
+            ServerEvent::ClientControlTerminal {
+                client_id,
+                target,
+                takeover,
+            } => self.control_terminal_client(client_id, target, takeover),
             ServerEvent::ClientAttachScroll {
                 client_id,
                 source,
@@ -2571,6 +2617,7 @@ impl HeadlessServer {
             }
             ServerEvent::ClientDetach { client_id } => {
                 info!(client_id, "client detached");
+                self.send_terminal_stream_detach_shutdown(client_id);
                 self.remove_client_and_resize_if_needed(client_id);
                 true
             }
@@ -4536,7 +4583,14 @@ next_tab = ""
     }
 
     fn connect_pending_terminal_client(server: &mut HeadlessServer, client_id: u64) {
-        let (writer, _control_rx, _render_rx) = test_client_writer();
+        let _control_rx = connect_pending_terminal_client_with_control_rx(server, client_id);
+    }
+
+    fn connect_pending_terminal_client_with_control_rx(
+        server: &mut HeadlessServer,
+        client_id: u64,
+    ) -> std::sync::mpsc::Receiver<Vec<u8>> {
+        let (writer, control_rx, _render_rx) = test_client_writer();
         assert!(server.handle_server_event(ServerEvent::ClientConnected {
             client_id,
             cols: 100,
@@ -4548,6 +4602,7 @@ next_tab = ""
             direct_attach_requested: true,
             writer,
         }));
+        control_rx
     }
 
     #[test]
@@ -4612,6 +4667,156 @@ next_tab = ""
     }
 
     #[test]
+    fn terminal_control_resolves_public_pane_id_and_takes_ownership() {
+        with_terminal_session_test_server(
+            |server, terminal_id, terminal_id_string, public_pane_id| {
+                connect_pending_terminal_client(server, 7);
+                assert!(
+                    server.handle_server_event(ServerEvent::ClientControlTerminal {
+                        client_id: 7,
+                        target: public_pane_id,
+                        takeover: false,
+                    })
+                );
+
+                assert!(matches!(
+                    server.clients.get(&7).map(|client| &client.mode),
+                    Some(ClientConnectionMode::TerminalAttach { terminal_id: attached })
+                        if attached == &terminal_id_string
+                ));
+                assert_eq!(
+                    server.terminal_attach_owners.get(&terminal_id_string),
+                    Some(&7)
+                );
+                assert!(server
+                    .app
+                    .state
+                    .direct_attach_resize_locks
+                    .contains(&terminal_id));
+            },
+        );
+    }
+
+    #[test]
+    fn terminal_control_rejects_second_controller_without_takeover() {
+        with_terminal_session_test_server(|server, _terminal_id, terminal_id_string, _| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+
+            connect_pending_terminal_client(server, 8);
+            assert!(
+                !server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 8,
+                    target: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+
+            assert!(server.clients.contains_key(&7));
+            assert!(!server.clients.contains_key(&8));
+            assert_eq!(
+                server.terminal_attach_owners.get(&terminal_id_string),
+                Some(&7)
+            );
+        });
+    }
+
+    #[test]
+    fn terminal_control_takeover_replaces_existing_controller() {
+        with_terminal_session_test_server(|server, _terminal_id, terminal_id_string, _| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+
+            connect_pending_terminal_client(server, 8);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 8,
+                    target: terminal_id_string.clone(),
+                    takeover: true,
+                })
+            );
+
+            assert!(!server.clients.contains_key(&7));
+            assert!(server.clients.contains_key(&8));
+            assert_eq!(
+                server.terminal_attach_owners.get(&terminal_id_string),
+                Some(&8)
+            );
+        });
+    }
+
+    #[test]
+    fn terminal_observe_can_coexist_with_terminal_control() {
+        with_terminal_session_test_server(|server, _terminal_id, terminal_id_string, _| {
+            connect_pending_terminal_client(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+
+            connect_pending_terminal_client(server, 8);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientObserveTerminal {
+                    client_id: 8,
+                    target: terminal_id_string.clone(),
+                })
+            );
+
+            assert_eq!(
+                server.terminal_attach_owners.get(&terminal_id_string),
+                Some(&7)
+            );
+            assert!(matches!(
+                server.clients.get(&8).map(|client| &client.mode),
+                Some(ClientConnectionMode::TerminalObserve { terminal_id })
+                    if terminal_id == &terminal_id_string
+            ));
+            assert_eq!(
+                terminal_stream_client_ids(&server.clients, &terminal_id_string).len(),
+                2
+            );
+        });
+    }
+
+    #[test]
+    fn terminal_control_detach_sends_shutdown_before_removal() {
+        with_terminal_session_test_server(|server, _terminal_id, terminal_id_string, _| {
+            let control_rx = connect_pending_terminal_client_with_control_rx(server, 7);
+            assert!(
+                server.handle_server_event(ServerEvent::ClientControlTerminal {
+                    client_id: 7,
+                    target: terminal_id_string.clone(),
+                    takeover: false,
+                })
+            );
+
+            assert!(server.handle_server_event(ServerEvent::ClientDetach { client_id: 7 }));
+
+            assert!(!server.clients.contains_key(&7));
+            assert!(!server
+                .terminal_attach_owners
+                .contains_key(&terminal_id_string));
+            let reason = read_server_shutdown_reason(control_rx.recv().expect("shutdown message"));
+            assert_eq!(reason, Some("detached".to_owned()));
+        });
+    }
+
+    #[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);