feat: add terminal session observe stream

9 files changed, 515 insertions, 8 deletions

diff --git a/docs/next/CHANGELOG.md b/docs/next/CHANGELOG.md
index 2df13b07..1236da08 100644
--- a/docs/next/CHANGELOG.md
+++ b/docs/next/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Added
 - Added `session.snapshot` to bootstrap client runtime state in one socket API response before subscribing to events.
 - Added `herdr api schema` to inspect the bundled socket API schema, with `--json` for the full JSON Schema document and `--output PATH` for file output.
+- Added `herdr terminal session observe` for read-only live ANSI terminal streams that bridge processes can consume as newline-delimited JSON.
 
 ### Changed
 - Bumped the client/server protocol version to 15 for socket API placement mutation event and response compatibility.
diff --git a/docs/next/website/src/content/docs/cli-reference.mdx b/docs/next/website/src/content/docs/cli-reference.mdx
index 361d09ac..d3f2f9c0 100644
--- a/docs/next/website/src/content/docs/cli-reference.mdx
+++ b/docs/next/website/src/content/docs/cli-reference.mdx
@@ -234,11 +234,17 @@ Use `pane send-text`, `pane send-keys`, `pane run`, and `terminal attach` for or
 
 ```bash
 herdr terminal attach <terminal_id> [--takeover]
+herdr terminal session observe <target> [--cols N] [--rows N]
 herdr terminal title set <title>
 herdr terminal title clear
 ```
 
 Detach from direct attach with `ctrl+b q`. Send literal `ctrl+b` with `ctrl+b ctrl+b`.
+`terminal session observe` opens a read-only live terminal stream for a pane,
+terminal, or agent target. It prints newline-delimited JSON `terminal.frame`
+records with base64-encoded ANSI bytes, then a `terminal.closed` record when
+the server closes the stream. Multiple observers can watch the same terminal
+without taking input, resize, scroll, or takeover authority.
 `terminal title clear` restores Herdr's default outer terminal window title.
 
 ## Waits
diff --git a/docs/next/website/src/content/docs/persistence-remote.mdx b/docs/next/website/src/content/docs/persistence-remote.mdx
index 80cf19d3..a2060832 100644
--- a/docs/next/website/src/content/docs/persistence-remote.mdx
+++ b/docs/next/website/src/content/docs/persistence-remote.mdx
@@ -121,6 +121,18 @@ Only one writable direct attach client owns input and resize for a terminal. Use
 herdr terminal attach term_abc123 --takeover
 ```
 
+For third-party bridges that only need rendered terminal bytes, use a read-only
+terminal session observer:
+
+```bash
+herdr terminal session observe w1:p1 --cols 120 --rows 40
+```
+
+It prints newline-delimited JSON `terminal.frame` records with base64 ANSI
+bytes, then a `terminal.closed` record when the server closes the stream.
+Multiple observers can watch the same terminal without taking input, resize,
+scroll, or takeover ownership.
+
 ## Single-process escape hatch
 
 Use `--no-session` to run Herdr without the background server/client split:
diff --git a/src/cli.rs b/src/cli.rs
index 0e0daf54..151286d1 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -20,6 +20,9 @@ mod tab;
 mod workspace;
 mod worktree;
 
+const TERMINAL_SESSION_OBSERVE_USAGE: &str =
+    "usage: herdr terminal session observe <target> [--cols N] [--rows N]";
+
 pub(crate) fn parse_env_assignment(raw: &str) -> Result<(String, String), String> {
     let Some((key, value)) = raw.split_once('=') else {
         return Err("env must use KEY=VALUE".into());
@@ -322,6 +325,7 @@ fn run_terminal_command(args: &[String]) -> std::io::Result<i32> {
 
     match subcommand {
         "attach" => terminal_attach(&args[1..]),
+        "session" => terminal_session(&args[1..]),
         "title" => terminal_title(&args[1..]),
         "help" | "--help" | "-h" => {
             print_terminal_help();
@@ -476,6 +480,86 @@ fn terminal_attach(args: &[String]) -> std::io::Result<i32> {
     Ok(0)
 }
 
+fn terminal_session(args: &[String]) -> std::io::Result<i32> {
+    match args.first().map(|arg| arg.as_str()) {
+        Some("observe") => terminal_session_observe(&args[1..]),
+        Some("help" | "--help" | "-h") => {
+            eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+            Ok(0)
+        }
+        _ => {
+            eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+            Ok(2)
+        }
+    }
+}
+
+fn terminal_session_observe(args: &[String]) -> std::io::Result<i32> {
+    if matches!(
+        args.first().map(|arg| arg.as_str()),
+        Some("help" | "--help" | "-h")
+    ) {
+        eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+        return Ok(0);
+    }
+    let Some(target) = args.first() else {
+        eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+        return Ok(2);
+    };
+
+    let mut cols = 120;
+    let mut rows = 40;
+    let mut i = 1;
+    while i < args.len() {
+        match args[i].as_str() {
+            "--cols" => {
+                let Some(value) = args.get(i + 1) else {
+                    eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+                    return Ok(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);
+                };
+                rows = parse_terminal_dimension(value, "--rows")?;
+                i += 2;
+            }
+            "help" | "--help" | "-h" => {
+                eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+                return Ok(0);
+            }
+            other => {
+                eprintln!("unknown terminal session observe option: {other}");
+                eprintln!("{TERMINAL_SESSION_OBSERVE_USAGE}");
+                return Ok(2);
+            }
+        }
+    }
+
+    crate::client::run_terminal_session_observe(target.clone(), cols, rows)?;
+    Ok(0)
+}
+
+fn parse_terminal_dimension(raw: &str, flag: &str) -> std::io::Result<u16> {
+    let parsed = raw.parse::<u16>().map_err(|_| {
+        std::io::Error::new(
+            std::io::ErrorKind::InvalidInput,
+            format!("{flag} must be an integer between 1 and {}", u16::MAX),
+        )
+    })?;
+    if parsed == 0 {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidInput,
+            format!("{flag} must be greater than 0"),
+        ));
+    }
+    Ok(parsed)
+}
+
 fn terminal_title(args: &[String]) -> std::io::Result<i32> {
     match args.first().map(|arg| arg.as_str()) {
         Some("set") => {
@@ -916,6 +1000,7 @@ fn print_config_help() {
 fn print_terminal_help() {
     eprintln!("herdr terminal commands:");
     eprintln!("  herdr terminal attach <terminal_id> [--takeover]");
+    eprintln!("  herdr terminal session observe <target> [--cols N] [--rows N]");
     eprintln!("  herdr terminal title set <title>");
     eprintln!("  herdr terminal title clear");
     eprintln!("  detach from direct attach with ctrl+b q; send literal ctrl+b with ctrl+b ctrl+b");
diff --git a/src/client/mod.rs b/src/client/mod.rs
index d9d73b95..98105781 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -20,6 +20,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
 
+use base64::Engine;
 use crossterm::event::{
     DisableBracketedPaste, DisableFocusChange, DisableMouseCapture, EnableBracketedPaste,
     EnableFocusChange, EnableMouseCapture,
@@ -814,6 +815,83 @@ 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<()> {
+    init_logging();
+
+    let socket_path = client_socket_path();
+    crate::logging::startup("client");
+    info!(path = %socket_path.display(), target = %target, cols, rows, "observing terminal session");
+
+    let mut stream = match crate::ipc::connect_local_stream(&socket_path) {
+        Ok(stream) => stream,
+        Err(err) => {
+            eprintln!("herdr: {}", ClientError::ConnectionFailed(err));
+            std::process::exit(1);
+        }
+    };
+
+    match do_handshake(
+        &mut stream,
+        cols,
+        rows,
+        0,
+        0,
+        RenderEncoding::TerminalAnsi,
+        true,
+    ) {
+        Ok(RenderEncoding::TerminalAnsi) => {}
+        Ok(encoding) => {
+            eprintln!(
+                "herdr: terminal session observe negotiated unsupported encoding {encoding:?}"
+            );
+            std::process::exit(1);
+        }
+        Err(err) => {
+            eprintln!("herdr: {err}");
+            std::process::exit(1);
+        }
+    }
+
+    write_to_server(&mut stream, &ClientMessage::ObserveTerminal { target })?;
+    stream.set_nonblocking(false)?;
+
+    let mut stdout = io::stdout().lock();
+    loop {
+        match protocol::read_message(&mut stream, MAX_GRAPHICS_FRAME_SIZE) {
+            Ok(ServerMessage::Terminal(frame)) => {
+                let encoded = base64::engine::general_purpose::STANDARD.encode(&frame.bytes);
+                let line = serde_json::json!({
+                    "type": "terminal.frame",
+                    "seq": frame.seq,
+                    "encoding": "ansi",
+                    "width": frame.width,
+                    "height": frame.height,
+                    "full": frame.full,
+                    "bytes": encoded,
+                });
+                serde_json::to_writer(&mut stdout, &line)?;
+                stdout.write_all(b"\n")?;
+                stdout.flush()?;
+            }
+            Ok(ServerMessage::ServerShutdown { reason }) => {
+                let line = serde_json::json!({
+                    "type": "terminal.closed",
+                    "reason": reason,
+                });
+                serde_json::to_writer(&mut stdout, &line)?;
+                stdout.write_all(b"\n")?;
+                stdout.flush()?;
+                return Ok(());
+            }
+            Ok(ServerMessage::Graphics { .. }) => {}
+            Ok(_) => {}
+            Err(protocol::FramingError::UnexpectedEof) => return Ok(()),
+            Err(err) => return Err(io::Error::other(err.to_string())),
+        }
+    }
+}
+
 fn run_client_with_mode(
     requested_encoding: RenderEncoding,
     attach_request: Option<(String, bool)>,
diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs
index ac21df19..e6ba5746 100644
--- a/src/protocol/wire.rs
+++ b/src/protocol/wire.rs
@@ -381,6 +381,12 @@ pub enum ClientMessage {
 
     /// Structured input events from platform clients that do not expose Unix-style raw bytes.
     InputEvents { events: Vec<ClientInputEvent> },
+
+    /// Switch this connection into read-only terminal observe mode.
+    ObserveTerminal {
+        /// Pane, terminal, or agent target to observe.
+        target: String,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -993,6 +999,12 @@ mod tests {
             6
         );
         assert_eq!(tag(&ClientMessage::InputEvents { events: Vec::new() }), 7);
+        assert_eq!(
+            tag(&ClientMessage::ObserveTerminal {
+                target: "w1:p1".to_owned(),
+            }),
+            8
+        );
     }
 
     #[test]
@@ -1121,6 +1133,17 @@ mod tests {
     }
 
     #[test]
+    fn client_observe_terminal_roundtrip() {
+        let msg = ClientMessage::ObserveTerminal {
+            target: "w1:p1".to_owned(),
+        };
+        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 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