fix(daemon): one-shot SendInput, displaced controllers close, observers get a budget

ClientMsg::SendInput (kind 55) writes to a pane's PTY without touching the
controlling subscriber, its epoch, or the size, answered by DaemonMsg::InputAck
(kind 51) or an Error for a missing or exited pane; PaneClient::send_input
wraps it. Both stay within protocol 5.

run_stream now polls its half of the socket and epoch-checks against the pane
before forwarding Input or Resize, so a controller displaced by a preempting
Attach stops writing into the shell and has its connection shut down instead of
half-open forwarding forever.

Each observer meters its queued Output through its own OutputGate; one that
lets 8 MiB pile up is pruned rather than growing daemon memory, while the
controller and the PTY never wait on it.

The pidfile reap guard accepts any legitimate daemon exe name (current exe,
tty7-app, tty7-server, tty7; .exe optional, case-insensitive on Windows), and
the agent-hooks console fast path matches tty7-server.exe and tty7.exe next to
tty7-app.exe.

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");
+}