merge: daemon review fixes — SendInput, displaced-controller close, observer budget

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv

crates/tty7-core/src/client/pane.rs

@@ -86,6 +86,20 @@ impl PaneClient {
         ClientMsg::Kill { pane_id }.encode(&mut stream)
     }
 
+    pub fn send_input(&self, pane_id: u64, bytes: &[u8]) -> io::Result<()> {
+        let mut stream = self.open()?;
+        ClientMsg::SendInput {
+            pane_id,
+            bytes: bytes.to_vec(),
+        }
+        .encode(&mut stream)?;
+        match DaemonMsg::read(&mut stream)? {
+            DaemonMsg::InputAck { .. } => Ok(()),
+            DaemonMsg::Error(message) => Err(io::Error::other(message)),
+            other => Err(unexpected_reply("SendInput", &other)),
+        }
+    }
+
     pub fn procs(&self, pane_id: u64) -> io::Result<PaneProcs> {
         let mut stream = self.open()?;
         ClientMsg::QueryProcs { pane_id }.encode(&mut stream)?;

crates/tty7-core/src/core/agent_hooks.rs

@@ -136,7 +136,7 @@ fn write_to_controlling_tty(bytes: &[u8]) -> bool {
             .iter()
             .find(|p| p.pid == pid)
             .and_then(|p| name_of(p.parent))
-            .is_some_and(|n| n == "tty7-app.exe")
+            .is_some_and(|n| is_tty7_host_exe(&n))
     });
 
     if let Some(pid) = shell {
@@ -152,6 +152,11 @@ fn write_to_controlling_tty(bytes: &[u8]) -> bool {
     any
 }
 
+#[cfg(any(not(unix), test))]
+fn is_tty7_host_exe(name: &str) -> bool {
+    matches!(name, "tty7-app.exe" | "tty7-server.exe" | "tty7.exe")
+}
+
 #[cfg(not(unix))]
 fn attach_and_write(pid: u32, bytes: &[u8]) -> bool {
     use windows_sys::Win32::System::Console::{AttachConsole, FreeConsole};
@@ -869,6 +874,16 @@ export default function (pi: ExtensionAPI) {{
 mod tests {
     use super::*;
 
+    #[test]
+    fn every_tty7_daemon_host_takes_the_console_fast_path() {
+        for name in ["tty7-app.exe", "tty7-server.exe", "tty7.exe"] {
+            assert!(is_tty7_host_exe(name), "{name} hosts tty7 shells");
+        }
+        for name in ["explorer.exe", "cmd.exe", "tty7", "tty7-app", "wt.exe"] {
+            assert!(!is_tty7_host_exe(name), "{name} is not a tty7 host process");
+        }
+    }
+
     #[test]
     fn hook_sequence_round_trips_through_the_daemon_parser() {
         use crate::core::cli_agent::{AgentEventKind, CLIAgent, parse_agent_event};

crates/tty7-core/src/daemon/pane.rs

@@ -398,6 +398,10 @@ impl OutputGate {
         self.drained.notify_all();
     }
 
+    pub(crate) fn queued_bytes(&self) -> i64 {
+        self.queued.load(Ordering::Relaxed)
+    }
+
     fn wait_below_high_water(&self) {
         if self.queued.load(Ordering::Relaxed) < Self::HIGH_WATER {
             return;
@@ -415,12 +419,20 @@ impl OutputGate {
     }
 }
 
+pub(crate) const OBSERVER_BUDGET: i64 = 8 * 1024 * 1024;
+
+struct Observer {
+    id: u64,
+    tx: Sender<DaemonMsg>,
+    gate: Arc<OutputGate>,
+}
+
 struct PaneState {
     id: u64,
     ring: ReplayRing,
     subscriber: Option<Sender<DaemonMsg>>,
     subscriber_epoch: u64,
-    observers: Vec<(u64, Sender<DaemonMsg>)>,
+    observers: Vec<Observer>,
     observer_seq: u64,
     cwd: Option<PathBuf>,
     shell: ShellState,
@@ -436,7 +448,25 @@ fn notify(st: &mut PaneState, msg: DaemonMsg) {
     if let Some(sub) = &st.subscriber {
         let _ = sub.send(msg.clone());
     }
-    st.observers.retain(|(_, tx)| tx.send(msg.clone()).is_ok());
+    st.observers.retain(|obs| obs.tx.send(msg.clone()).is_ok());
+}
+
+fn fan_out_output(st: &mut PaneState, bytes: &[u8], gate: &OutputGate) {
+    if let Some(sub) = &st.subscriber {
+        if sub.send(DaemonMsg::Output(bytes.to_vec())).is_ok() {
+            gate.add(bytes.len());
+        }
+    }
+    st.observers.retain(|obs| {
+        if obs.gate.queued_bytes() + bytes.len() as i64 > OBSERVER_BUDGET {
+            return false;
+        }
+        if obs.tx.send(DaemonMsg::Output(bytes.to_vec())).is_err() {
+            return false;
+        }
+        obs.gate.add(bytes.len());
+        true
+    });
 }
 
 enum PaneBackend {
@@ -840,14 +870,7 @@ impl DaemonPane {
                             let mut st = state.lock().unwrap();
                             let facts_before = may_change_facts.then(|| observed_facts(&st));
                             st.ring.append(bytes);
-                            if let Some(sub) = &st.subscriber {
-                                if sub.send(DaemonMsg::Output(bytes.to_vec())).is_ok() {
-                                    gate.add(n);
-                                }
-                            }
-                            st.observers.retain(|(_, tx)| {
-                                tx.send(DaemonMsg::Output(bytes.to_vec())).is_ok()
-                            });
+                            fan_out_output(&mut st, bytes, &gate);
                             apply_signals(&mut st, signals);
                             if let Some(remote) = remote {
                                 apply_remote_context(&mut st, remote);
@@ -905,14 +928,18 @@ impl DaemonPane {
         !st.alive && st.subscriber.is_none()
     }
 
-    pub fn observe(&self, observer: Sender<DaemonMsg>) -> u64 {
+    pub fn observe(&self, observer: Sender<DaemonMsg>, gate: Arc<OutputGate>) -> u64 {
         let mut st = self.state.lock().unwrap();
-        observe_subscriber(&mut st, observer)
+        observe_subscriber(&mut st, observer, gate)
     }
 
     pub fn unobserve(&self, observer_id: u64) {
         let mut st = self.state.lock().unwrap();
-        st.observers.retain(|(id, _)| *id != observer_id);
+        st.observers.retain(|obs| obs.id != observer_id);
+    }
+
+    pub fn controls(&self, epoch: u64) -> bool {
+        self.state.lock().unwrap().subscriber_epoch == epoch
     }
 
     pub fn agent_state(&self) -> Option<crate::daemon::control::PaneAgentState> {
@@ -955,7 +982,7 @@ impl DaemonPane {
             let mut st = self.state.lock().unwrap();
             st.ring.resize(size);
             st.observers
-                .retain(|(_, tx)| tx.send(DaemonMsg::Size(size)).is_ok());
+                .retain(|obs| obs.tx.send(DaemonMsg::Size(size)).is_ok());
         }
         match &self.backend {
             PaneBackend::Pty(p) => {
@@ -967,7 +994,6 @@ impl DaemonPane {
         }
     }
 
-    #[allow(dead_code)]
     pub fn alive(&self) -> bool {
         self.state.lock().unwrap().alive
     }
@@ -1301,10 +1327,14 @@ fn attach_subscriber(st: &mut PaneState, subscriber: Sender<DaemonMsg>) -> u64 {
     st.subscriber_epoch
 }
 
-fn observe_subscriber(st: &mut PaneState, observer: Sender<DaemonMsg>) -> u64 {
+fn observe_subscriber(st: &mut PaneState, observer: Sender<DaemonMsg>, gate: Arc<OutputGate>) -> u64 {
     st.observer_seq += 1;
     replay_state(st, &observer);
-    st.observers.push((st.observer_seq, observer));
+    st.observers.push(Observer {
+        id: st.observer_seq,
+        tx: observer,
+        gate,
+    });
     st.observer_seq
 }
 
@@ -2979,7 +3009,7 @@ mod tests {
         drain(&controller_rx);
 
         let (observer_tx, observer_rx) = mpsc::channel();
-        let id = observe_subscriber(&mut st, observer_tx);
+        let id = observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()));
         assert_eq!(st.subscriber_epoch, epoch, "observing must not bump the controller epoch");
         assert!(st.subscriber.is_some(), "the controller keeps its seat");
 
@@ -2995,7 +3025,7 @@ mod tests {
         assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick"));
         assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick"));
 
-        st.observers.retain(|(oid, _)| *oid != id);
+        st.observers.retain(|obs| obs.id != id);
         notify(&mut st, DaemonMsg::Output(b"tock".to_vec()));
         assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Output(_))));
         assert!(observer_rx.try_recv().is_err(), "a departed observer hears nothing");
@@ -3010,7 +3040,7 @@ mod tests {
         drain(&first_rx);
 
         let (observer_tx, observer_rx) = mpsc::channel();
-        observe_subscriber(&mut st, observer_tx);
+        observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()));
         drain(&observer_rx);
 
         let (second_tx, second_rx) = mpsc::channel();
@@ -3042,18 +3072,92 @@ mod tests {
     fn a_gone_observer_is_pruned_on_the_next_broadcast() {
         let mut st = test_state(true);
         let (observer_tx, observer_rx) = mpsc::channel();
-        observe_subscriber(&mut st, observer_tx);
+        observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()));
         drop(observer_rx);
 
         notify(&mut st, DaemonMsg::Output(b"x".to_vec()));
         assert!(st.observers.is_empty(), "a dead observer must not accumulate");
     }
 
+    #[test]
+    fn a_stalled_observer_is_dropped_at_its_budget_while_the_controller_streams_on() {
+        let mut st = test_state(true);
+        let (controller_tx, controller_rx) = mpsc::channel();
+        attach_subscriber(&mut st, controller_tx);
+        drain(&controller_rx);
+
+        let (observer_tx, observer_rx) = mpsc::channel();
+        observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()));
+        drain(&observer_rx);
+
+        let pane_gate = OutputGate::new();
+        let chunk = vec![b'x'; 1024 * 1024];
+        let sends = (OBSERVER_BUDGET / chunk.len() as i64) as usize + 2;
+        for _ in 0..sends {
+            fan_out_output(&mut st, &chunk, &pane_gate);
+            pane_gate.sub(chunk.len());
+        }
+
+        assert!(
+            st.observers.is_empty(),
+            "an observer past its budget must be pruned"
+        );
+        let mut controller_bytes = 0usize;
+        while let Ok(DaemonMsg::Output(b)) = controller_rx.try_recv() {
+            controller_bytes += b.len();
+        }
+        assert_eq!(
+            controller_bytes,
+            sends * chunk.len(),
+            "the controller stream must stay complete"
+        );
+        let mut observer_bytes = 0i64;
+        while let Ok(DaemonMsg::Output(b)) = observer_rx.try_recv() {
+            observer_bytes += b.len() as i64;
+        }
+        assert!(
+            observer_bytes <= OBSERVER_BUDGET,
+            "a stalled observer must never hold more than its budget, held {observer_bytes}"
+        );
+    }
+
+    #[test]
+    fn a_draining_observer_under_the_cap_stays_subscribed() {
+        let mut st = test_state(true);
+        let (observer_tx, observer_rx) = mpsc::channel();
+        let observer_gate = Arc::new(OutputGate::new());
+        observe_subscriber(&mut st, observer_tx, observer_gate.clone());
+        drain(&observer_rx);
+
+        let pane_gate = OutputGate::new();
+        let chunk = vec![b'y'; 1024 * 1024];
+        let sends = (OBSERVER_BUDGET / chunk.len() as i64) as usize * 3;
+        let mut got = 0usize;
+        for _ in 0..sends {
+            fan_out_output(&mut st, &chunk, &pane_gate);
+            pane_gate.sub(chunk.len());
+            while let Ok(DaemonMsg::Output(b)) = observer_rx.try_recv() {
+                observer_gate.sub(b.len());
+                got += b.len();
+            }
+        }
+        assert_eq!(
+            st.observers.len(),
+            1,
+            "an observer that keeps draining must stay subscribed"
+        );
+        assert_eq!(got, sends * chunk.len(), "and must miss no bytes");
+    }
+
     #[test]
     fn death_notifies_observers_but_only_controllers_defer_the_reap() {
         let with_observer_only = Arc::new(Mutex::new(test_state(true)));
         let (observer_tx, observer_rx) = mpsc::channel();
-        observe_subscriber(&mut with_observer_only.lock().unwrap(), observer_tx);
+        observe_subscriber(
+            &mut with_observer_only.lock().unwrap(),
+            observer_tx,
+            Arc::new(OutputGate::new()),
+        );
         drain(&observer_rx);
         let (dead_tx, dead_rx) = mpsc::channel();
         DeathReporter::new(move || dead_tx.send(()).unwrap())
@@ -3070,7 +3174,7 @@ mod tests {
         {
             let mut st = with_both.lock().unwrap();
             attach_subscriber(&mut st, controller_tx);
-            observe_subscriber(&mut st, observer_tx);
+            observe_subscriber(&mut st, observer_tx, Arc::new(OutputGate::new()));
         }
         drain(&controller_rx);
         drain(&observer_rx);

crates/tty7-core/src/daemon/protocol.rs

@@ -584,6 +584,10 @@ pub enum ClientMsg {
         size: WinSize,
     },
     Input(Vec<u8>),
+    SendInput {
+        pane_id: u64,
+        bytes: Vec<u8>,
+    },
     Resize(WinSize),
     Detach,
     Kill {
@@ -656,6 +660,9 @@ pub enum DaemonMsg {
         code: Option<i32>,
     },
     PaneList(Vec<PaneInfo>),
+    InputAck {
+        pane_id: u64,
+    },
     RemoteContext(Option<RemoteContext>),
     Agent(Option<crate::core::cli_agent::CLIAgent>),
     AgentStatus(Option<crate::core::cli_agent::AgentSessionState>),
@@ -711,6 +718,7 @@ mod kind {
     pub const ON_WORKSPACE: u8 = 52;
     pub const SPAWN_OWNED: u8 = 53;
     pub const OBSERVE: u8 = 54;
+    pub const SEND_INPUT: u8 = 55;
 
     pub const SPAWNED: u8 = 1;
     pub const SNAPSHOT: u8 = 2;
@@ -736,6 +744,7 @@ mod kind {
     pub const AGENT_STATUS: u8 = 22;
     pub const VERSION_REPLY: u8 = 40;
     pub const PROCS: u8 = 50;
+    pub const INPUT_ACK: u8 = 51;
 }
 
 pub fn write_frame<W: Write>(w: &mut W, kind: u8, payload: &[u8]) -> io::Result<()> {
@@ -860,6 +869,9 @@ impl ClientMsg {
                 write_frame(w, kind::OBSERVE, &to_json(&(pane_id, size))?)
             }
             ClientMsg::Input(bytes) => write_frame(w, kind::INPUT, bytes),
+            ClientMsg::SendInput { pane_id, bytes } => {
+                write_frame(w, kind::SEND_INPUT, &to_json(&(pane_id, bytes))?)
+            }
             ClientMsg::Resize(size) => write_frame(w, kind::RESIZE, &to_json(size)?),
             ClientMsg::Detach => write_frame(w, kind::DETACH, &[]),
             ClientMsg::Kill { pane_id } => write_frame(w, kind::KILL, &to_json(pane_id)?),
@@ -963,6 +975,10 @@ impl ClientMsg {
                 ClientMsg::Observe { pane_id, size }
             }
             kind::INPUT => ClientMsg::Input(payload),
+            kind::SEND_INPUT => {
+                let (pane_id, bytes) = from_json(&payload)?;
+                ClientMsg::SendInput { pane_id, bytes }
+            }
             kind::RESIZE => ClientMsg::Resize(from_json(&payload)?),
             kind::DETACH => ClientMsg::Detach,
             kind::KILL => ClientMsg::Kill {
@@ -1050,6 +1066,7 @@ impl DaemonMsg {
             } => write_frame(w, kind::PROMPT, &to_json(&(active, at_prompt, last_exit))?),
             DaemonMsg::Exited { code } => write_frame(w, kind::EXITED, &to_json(code)?),
             DaemonMsg::PaneList(list) => write_frame(w, kind::PANE_LIST, &to_json(list)?),
+            DaemonMsg::InputAck { pane_id } => write_frame(w, kind::INPUT_ACK, &to_json(pane_id)?),
             DaemonMsg::RemoteContext(remote) => {
                 write_frame(w, kind::REMOTE_CONTEXT, &to_json(remote)?)
             }
@@ -1108,6 +1125,9 @@ impl DaemonMsg {
                 code: from_json(&payload)?,
             },
             kind::PANE_LIST => DaemonMsg::PaneList(from_json(&payload)?),
+            kind::INPUT_ACK => DaemonMsg::InputAck {
+                pane_id: from_json(&payload)?,
+            },
             kind::REMOTE_CONTEXT => DaemonMsg::RemoteContext(from_json(&payload)?),
             kind::AGENT => DaemonMsg::Agent(from_json(&payload)?),
             kind::AGENT_STATUS => DaemonMsg::AgentStatus(from_json(&payload)?),
@@ -1269,6 +1289,14 @@ mod tests {
                 size: SIZE,
             },
             ClientMsg::Input(vec![0x1b, b'[', b'A', 0, 255]),
+            ClientMsg::SendInput {
+                pane_id: 42,
+                bytes: vec![b'l', b's', b'\r', 0, 255],
+            },
+            ClientMsg::SendInput {
+                pane_id: 7,
+                bytes: Vec::new(),
+            },
             ClientMsg::Resize(SIZE),
             ClientMsg::Detach,
             ClientMsg::Kill { pane_id: 7 },
@@ -1405,6 +1433,7 @@ mod tests {
                     owner: Some("ffe038d0-9ad6-40c0-815d-1fcc43c17ec0".into()),
                 },
             ]),
+            DaemonMsg::InputAck { pane_id: 42 },
             DaemonMsg::RemoteContext(Some(RemoteContext {
                 kind: RemoteKind::Ssh,
                 argv: vec!["ssh".into(), "-p".into(), "2222".into(), "dev".into()],

crates/tty7-core/src/daemon/server.rs

@@ -556,6 +556,21 @@ fn handle_conn(stream: Stream, registry: Arc<Registry>) -> anyhow::Result<()> {
             Ok(())
         }
 
+        ClientMsg::SendInput { pane_id, bytes } => {
+            let mut w = write_stream;
+            match registry.get(pane_id) {
+                Some(pane) if pane.alive() => {
+                    pane.write_input(&bytes);
+                    DaemonMsg::InputAck { pane_id }.encode(&mut w)?;
+                }
+                Some(_) => {
+                    DaemonMsg::Error(format!("pane {pane_id} is not running")).encode(&mut w)?
+                }
+                None => DaemonMsg::Error(format!("no such pane {pane_id}")).encode(&mut w)?,
+            }
+            Ok(())
+        }
+
         ClientMsg::ListForwards { pane_id } => {
             let mut w = write_stream;
             let list = crate::daemon::ssh::SshManager::global().list_forwards(pane_id);
@@ -618,12 +633,9 @@ fn stream_observer(
 ) -> anyhow::Result<()> {
     let (tx, rx) = mpsc::channel::<DaemonMsg>();
     let refusals = tx.clone();
-    let observer_id = pane.observe(tx);
-    let writer = spawn_writer(
-        rx,
-        write_stream,
-        Arc::new(crate::daemon::pane::OutputGate::new()),
-    );
+    let gate = Arc::new(crate::daemon::pane::OutputGate::new());
+    let observer_id = pane.observe(tx, gate.clone());
+    let writer = spawn_writer(rx, write_stream, gate);
 
     observe_loop(&mut read_stream, &refusals);
 
@@ -651,6 +663,8 @@ fn observe_loop<R: std::io::Read>(read_stream: &mut R, refusals: &mpsc::Sender<D
     }
 }
 
+const CONTROL_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(200);
+
 fn run_stream(
     pane: Arc<DaemonPane>,
     id: u64,
@@ -660,32 +674,81 @@ fn run_stream(
     write_stream: Stream,
     registry: Arc<Registry>,
 ) -> anyhow::Result<()> {
+    use std::io::Read as _;
+
     let writer = spawn_writer(rx, write_stream, pane.gate());
+    let _ = read_stream.set_read_timeout(Some(CONTROL_POLL_INTERVAL));
 
     let mut killed = false;
-    loop {
-        match ClientMsg::read(&mut read_stream) {
-            Ok(ClientMsg::Input(bytes)) => pane.write_input(&bytes),
-            Ok(ClientMsg::Resize(size)) => pane.resize(size),
-            Ok(ClientMsg::AuthResponse {
-                request_id,
-                response,
-            }) => pane.deliver_auth_response(request_id, response),
-            Ok(ClientMsg::Detach) => break,
-            Ok(ClientMsg::Kill { pane_id }) => {
-                if pane_id == id {
-                    killed = true;
-                    break;
-                } else if let Some(other) = registry.remove(pane_id) {
-                    other.kill();
+    let mut displaced = false;
+    let mut pending: Vec<u8> = Vec::new();
+    let mut chunk = [0u8; 65536];
+    'conn: loop {
+        if !pane.controls(epoch) {
+            displaced = true;
+            break;
+        }
+        loop {
+            let (kind, payload) = match crate::daemon::protocol::take_frame(&mut pending) {
+                Ok(Some(frame)) => frame,
+                Ok(None) => break,
+                Err(_) => break 'conn,
+            };
+            let Ok(msg) = ClientMsg::from_frame(kind, payload) else {
+                break 'conn;
+            };
+            match msg {
+                ClientMsg::Input(bytes) => {
+                    if !pane.controls(epoch) {
+                        displaced = true;
+                        break 'conn;
+                    }
+                    pane.write_input(&bytes);
                 }
+                ClientMsg::Resize(size) => {
+                    if !pane.controls(epoch) {
+                        displaced = true;
+                        break 'conn;
+                    }
+                    pane.resize(size);
+                }
+                ClientMsg::AuthResponse {
+                    request_id,
+                    response,
+                } => pane.deliver_auth_response(request_id, response),
+                ClientMsg::Detach => break 'conn,
+                ClientMsg::Kill { pane_id } => {
+                    if pane_id == id {
+                        killed = true;
+                        break 'conn;
+                    } else if let Some(other) = registry.remove(pane_id) {
+                        other.kill();
+                    }
+                }
+                _ => {}
+            }
+        }
+        match read_stream.read(&mut chunk) {
+            Ok(0) => break,
+            Ok(n) => pending.extend_from_slice(&chunk[..n]),
+            Err(e)
+                if matches!(
+                    e.kind(),
+                    std::io::ErrorKind::WouldBlock
+                        | std::io::ErrorKind::TimedOut
+                        | std::io::ErrorKind::Interrupted
+                ) =>
+            {
+                continue;
             }
-            Ok(_) => {}
             Err(_) => break,
         }
     }
 
     let reclaimable = pane.detach(epoch);
+    if displaced {
+        let _ = read_stream.shutdown(std::net::Shutdown::Both);
+    }
     let _ = writer.join();
 
     if killed {

crates/tty7-core/src/daemon/spawn.rs

@@ -49,6 +49,35 @@ enum VersionProbe {
     Unresponsive,
 }
 
+const DAEMON_EXE_STEMS: [&str; 3] = ["tty7-app", "tty7-server", "tty7"];
+
+fn strip_exe_suffix(name: &str) -> &str {
+    match name.len().checked_sub(4) {
+        Some(i) if name.is_char_boundary(i) && name[i..].eq_ignore_ascii_case(".exe") => {
+            &name[..i]
+        }
+        _ => name,
+    }
+}
+
+fn exe_names_equal(a: &str, b: &str) -> bool {
+    let a = strip_exe_suffix(a);
+    let b = strip_exe_suffix(b);
+    if cfg!(windows) {
+        a.eq_ignore_ascii_case(b)
+    } else {
+        a == b
+    }
+}
+
+fn is_reapable_daemon_name(name: &str) -> bool {
+    let own = std::env::current_exe()
+        .ok()
+        .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()));
+    own.as_deref().is_some_and(|own| exe_names_equal(own, name))
+        || DAEMON_EXE_STEMS.iter().any(|stem| exe_names_equal(stem, name))
+}
+
 pub fn ensure_running() -> anyhow::Result<()> {
     if let Ok(mut stream) = transport::connect() {
         match query_daemon_version(&mut stream) {
@@ -174,7 +203,7 @@ fn reap_recorded_daemon() {
         pidfile::remove();
         return;
     }
-    if process_matches_own_exe(pid as libc::pid_t) {
+    if process_matches_daemon_exe(pid as libc::pid_t) {
         log::warn!("reaping unreachable daemon (pid {pid}); its sessions will be hung up");
         reap_process(pid as libc::pid_t);
     }
@@ -182,12 +211,10 @@ fn reap_recorded_daemon() {
 }
 
 #[cfg(any(target_os = "macos", target_os = "linux"))]
-fn process_matches_own_exe(pid: libc::pid_t) -> bool {
-    let ours = std::env::current_exe()
-        .ok()
-        .and_then(|p| p.file_name().map(|n| n.to_os_string()));
-    let theirs = process_path(pid).and_then(|p| p.file_name().map(|n| n.to_os_string()));
-    matches!((ours, theirs), (Some(a), Some(b)) if a == b)
+fn process_matches_daemon_exe(pid: libc::pid_t) -> bool {
+    process_path(pid)
+        .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()))
+        .is_some_and(|name| is_reapable_daemon_name(&name))
 }
 
 #[cfg(any(target_os = "macos", target_os = "linux"))]
@@ -228,14 +255,10 @@ fn reap_recorded_daemon() {
         return;
     }
     let procs = winproc::snapshot();
-    let ours = std::env::current_exe()
-        .ok()
-        .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()));
     let matches = procs
         .iter()
         .find(|p| p.pid == pid)
-        .zip(ours)
-        .is_some_and(|(entry, name)| entry.name.eq_ignore_ascii_case(&name));
+        .is_some_and(|entry| is_reapable_daemon_name(&entry.name));
     if matches {
         log::warn!("reaping unreachable daemon (pid {pid}); its sessions will be hung up");
         for descendant in winproc::descendants(&procs, pid) {
@@ -345,6 +368,74 @@ fn detach(cmd: &mut Command) {
     cmd.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW);
 }
 
+#[cfg(test)]
+mod exe_name_tests {
+    use super::*;
+
+    #[test]
+    fn every_legitimate_daemon_name_is_reapable_with_and_without_exe() {
+        for name in [
+            "tty7-app",
+            "tty7-server",
+            "tty7",
+            "tty7-app.exe",
+            "tty7-server.exe",
+            "tty7.exe",
+        ] {
+            assert!(is_reapable_daemon_name(name), "{name} is a daemon of ours");
+        }
+    }
+
+    #[test]
+    fn the_current_executable_name_remains_reapable() {
+        let own = std::env::current_exe()
+            .unwrap()
+            .file_name()
+            .unwrap()
+            .to_string_lossy()
+            .into_owned();
+        assert!(
+            is_reapable_daemon_name(&own),
+            "{own} launched this process and must stay in the set"
+        );
+    }
+
+    #[test]
+    fn foreign_process_names_are_never_reapable() {
+        for name in [
+            "explorer.exe",
+            "sleep",
+            "tty7d",
+            "nottty7",
+            "tty7-app2",
+            "tty7.",
+            "",
+        ] {
+            assert!(
+                !is_reapable_daemon_name(name),
+                "{name:?} must be protected from the reap"
+            );
+        }
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn windows_matches_daemon_names_case_insensitively() {
+        assert!(is_reapable_daemon_name("TTY7-APP.EXE"));
+        assert!(is_reapable_daemon_name("Tty7-Server"));
+        assert!(is_reapable_daemon_name("TTY7"));
+    }
+
+    #[test]
+    fn strip_exe_suffix_only_strips_a_trailing_exe() {
+        assert_eq!(strip_exe_suffix("tty7-app.exe"), "tty7-app");
+        assert_eq!(strip_exe_suffix("tty7-app.EXE"), "tty7-app");
+        assert_eq!(strip_exe_suffix("tty7-app"), "tty7-app");
+        assert_eq!(strip_exe_suffix(".exe"), "");
+        assert_eq!(strip_exe_suffix("exe"), "exe");
+    }
+}
+
 #[cfg(all(test, unix))]
 mod tests {
     use super::*;
@@ -387,8 +478,8 @@ mod tests {
             std::thread::sleep(Duration::from_millis(10));
         }
         assert!(
-            !process_matches_own_exe(pid),
-            "sleep must not match the test binary; matching here would mean the reap could kill it"
+            !process_matches_daemon_exe(pid),
+            "sleep must not match any daemon name; matching here would mean the reap could kill it"
         );
 
         let _ = child.kill();

crates/tty7-server/tests/pane_send_input.rs

@@ -0,0 +1,267 @@
+use std::io;
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+use tty7_core::client::PaneClient;
+use tty7_core::daemon::protocol::{DaemonMsg, ShellSpec, WinSize};
+
+const READY_WITHIN: Duration = Duration::from_secs(30);
+const STREAM_WITHIN: Duration = Duration::from_secs(30);
+const EOF_WITHIN: Duration = Duration::from_secs(10);
+
+struct Daemon {
+    child: Child,
+    dir: tempfile::TempDir,
+}
+
+impl Daemon {
+    fn start() -> Daemon {
+        let dir = tempfile::TempDir::new().unwrap();
+        let child = Command::new(env!("CARGO_BIN_EXE_tty7-server"))
+            .arg("--daemon")
+            .arg("--config-dir")
+            .arg(dir.path())
+            .env("TTY7_DATA_DIR", dir.path())
+            .env("TTY7_CONTROL_SOCK", dir.path().join("control.sock"))
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .expect("start tty7-server --daemon");
+        let daemon = Daemon { child, dir };
+        daemon.await_ready();
+        daemon
+    }
+
+    fn pane_endpoint(&self) -> PathBuf {
+        let file = if cfg!(windows) {
+            "daemon.port"
+        } else {
+            "daemon.sock"
+        };
+        self.dir.path().join(file)
+    }
+
+    fn panes(&self) -> PaneClient {
+        PaneClient::at(self.pane_endpoint())
+    }
+
+    fn await_ready(&self) {
+        let deadline = Instant::now() + READY_WITHIN;
+        loop {
+            if self.panes().version().is_ok() {
+                return;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "tty7-server did not open its pane endpoint within {READY_WITHIN:?}"
+            );
+            std::thread::sleep(Duration::from_millis(50));
+        }
+    }
+}
+
+impl Drop for Daemon {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+fn size() -> WinSize {
+    WinSize {
+        cols: 100,
+        rows: 30,
+        cell_w: 8,
+        cell_h: 16,
+    }
+}
+
+fn one_shot_shell(command: &str) -> ShellSpec {
+    if cfg!(windows) {
+        ShellSpec {
+            program: "cmd.exe".into(),
+            args: vec!["/d".into(), "/c".into(), command.into()],
+            args_are_tty7_defaults: false,
+        }
+    } else {
+        ShellSpec {
+            program: "/bin/sh".into(),
+            args: vec!["-c".into(), command.into()],
+            args_are_tty7_defaults: false,
+        }
+    }
+}
+
+fn interactive_shell() -> ShellSpec {
+    if cfg!(windows) {
+        ShellSpec {
+            program: "cmd.exe".into(),
+            args: vec!["/d".into()],
+            args_are_tty7_defaults: false,
+        }
+    } else {
+        ShellSpec {
+            program: "/bin/sh".into(),
+            args: Vec::new(),
+            args_are_tty7_defaults: false,
+        }
+    }
+}
+
+fn windows_contain(haystack: &[u8], needle: &[u8]) -> bool {
+    haystack.windows(needle.len()).any(|w| w == needle)
+}
+
+fn collect_until(session: &mut tty7_core::client::PaneSession, marker: &[u8]) -> Vec<u8> {
+    let mut seen: Vec<u8> = Vec::new();
+    loop {
+        match session.recv() {
+            Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => {
+                seen.extend_from_slice(&bytes);
+                if windows_contain(&seen, marker) {
+                    return seen;
+                }
+            }
+            Ok(DaemonMsg::Exited { code }) => panic!(
+                "pane exited ({code:?}) before {:?} appeared; saw {:?}",
+                String::from_utf8_lossy(marker),
+                String::from_utf8_lossy(&seen)
+            ),
+            Ok(_) => {}
+            Err(e) => panic!(
+                "pane stream ended early: {e}; saw {:?}",
+                String::from_utf8_lossy(&seen)
+            ),
+        }
+    }
+}
+
+#[test]
+fn send_input_reaches_the_shell_without_displacing_the_controller() {
+    let daemon = Daemon::start();
+    let panes = daemon.panes();
+    let mut session = panes
+        .spawn(None, size(), Some(interactive_shell()), None, None)
+        .expect("spawn an interactive pane");
+    let pane_id = session.pane_id();
+    session
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+
+    panes
+        .send_input(pane_id, b"echo tty7_send_oneshot\r")
+        .expect("one-shot input is acknowledged");
+    collect_until(&mut session, b"tty7_send_oneshot");
+
+    session
+        .input(b"echo tty7_still_controller\r")
+        .expect("the controller keeps its seat");
+    collect_until(&mut session, b"tty7_still_controller");
+
+    session.kill().expect("kill the pane");
+}
+
+#[test]
+fn send_input_to_a_missing_pane_answers_an_error() {
+    let daemon = Daemon::start();
+    let err = daemon
+        .panes()
+        .send_input(u64::MAX, b"echo lost\r")
+        .expect_err("input into a pane that never existed must fail");
+    assert!(
+        err.to_string().contains("no such pane"),
+        "the refusal was {err}"
+    );
+}
+
+#[test]
+fn send_input_to_a_dead_pane_answers_an_error() {
+    let daemon = Daemon::start();
+    let panes = daemon.panes();
+    let mut session = panes
+        .spawn(None, size(), Some(one_shot_shell("exit 0")), None, None)
+        .expect("spawn a one-shot pane");
+    let pane_id = session.pane_id();
+    session
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+    loop {
+        match session.recv() {
+            Ok(DaemonMsg::Exited { .. }) => break,
+            Ok(_) => {}
+            Err(e) => panic!("pane stream ended before Exited: {e}"),
+        }
+    }
+
+    let err = panes
+        .send_input(pane_id, b"echo too_late\r")
+        .expect_err("input into an exited pane must fail");
+    assert!(
+        err.to_string().contains("not running"),
+        "the refusal was {err}"
+    );
+}
+
+#[test]
+fn a_preempting_attach_closes_the_displaced_controller_and_drops_its_input() {
+    let daemon = Daemon::start();
+    let panes = daemon.panes();
+    let mut first = panes
+        .spawn(None, size(), Some(interactive_shell()), None, None)
+        .expect("spawn an interactive pane");
+    let pane_id = first.pane_id();
+    first
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+    first
+        .input(b"echo tty7_first_seated\r")
+        .expect("the first controller types");
+    collect_until(&mut first, b"tty7_first_seated");
+
+    let mut second = panes.attach(pane_id, size()).expect("preempting attach");
+    second
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the second stream reads");
+
+    let _ = first.input(b"echo tty7_stale_input\r");
+
+    first
+        .set_recv_timeout(Some(EOF_WITHIN))
+        .expect("bound the displaced reads");
+    let deadline = Instant::now() + EOF_WITHIN + Duration::from_secs(5);
+    let eof = loop {
+        match first.recv() {
+            Ok(_) => {
+                assert!(
+                    Instant::now() < deadline,
+                    "the displaced controller's stream never ended"
+                );
+            }
+            Err(e) => break e,
+        }
+    };
+    assert_ne!(
+        eof.kind(),
+        io::ErrorKind::TimedOut,
+        "the displaced connection must be closed, not left dangling: {eof}"
+    );
+    assert_ne!(
+        eof.kind(),
+        io::ErrorKind::WouldBlock,
+        "the displaced connection must be closed, not left dangling: {eof}"
+    );
+
+    second
+        .input(b"echo tty7_second_alive\r")
+        .expect("the new controller types");
+    let seen = collect_until(&mut second, b"tty7_second_alive");
+    assert!(
+        !windows_contain(&seen, b"tty7_stale_input"),
+        "input from the displaced controller must never reach the shell: {:?}",
+        String::from_utf8_lossy(&seen)
+    );
+
+    second.kill().expect("kill the pane");
+}