fix(daemon): wake displaced controllers without polling, keep busy links usable

run_stream polled pane.controls(epoch) every 200ms behind a read timeout, so
every attached pane woke its thread five times a second just to notice a
handover that may never come. The writer already learns of the handover the
instant it happens — its channel closes — so it now shuts the connection's read
side down on its way out, and the reader goes back to a plain blocking read.

SshManager::routes() reported a link as disconnected whenever its slot's
try_lock failed, which is precisely when the link is in use. The CLI's
-m <machine> refuses to route over a link it is told is down, so an actively
used connection would intermittently fail. Busy now reads as connected, matching
how SshConnection::is_alive resolves the same contention.

Also: the exit-code probe takes the child lock with try_lock, since Drop holds
it across a blocking wait(), and its window drops from 2s to 500ms — it only
needs to cover the race between pty EOF and the child becoming reapable, and
everything past that is a pane that looks frozen to every client. Observer
budget now covers status traffic and the initial replay, not just output.
Uptime is anchored where the control listener opens so a GUI-hosted server
does not report itself as freshly started.

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

@@ -421,6 +421,15 @@ impl OutputGate {
 
 pub(crate) const OBSERVER_BUDGET: i64 = 8 * 1024 * 1024;
 
+/// How long the death path will chase a child's exit status before giving up
+/// and reporting `Exited { code: None }`. The pty hits EOF when the last slave
+/// fd closes, which can land a few milliseconds ahead of the child becoming
+/// reapable — that race is all this window exists to cover. Everything past it
+/// is a pane that looks frozen to every attached client, so it stays short.
+const EXIT_CODE_PROBE_WINDOW: Duration = Duration::from_millis(500);
+
+const EXIT_CODE_PROBE_INTERVAL: Duration = Duration::from_millis(10);
+
 struct Observer {
     id: u64,
     tx: Sender<DaemonMsg>,
@@ -448,7 +457,13 @@ fn notify(st: &mut PaneState, msg: DaemonMsg) {
     if let Some(sub) = &st.subscriber {
         let _ = sub.send(msg.clone());
     }
-    st.observers.retain(|obs| obs.tx.send(msg.clone()).is_ok());
+    // Status traffic is charged the same budget as output. These messages are
+    // small, but an agent pane emits AgentStatus often enough that an observer
+    // which stopped draining would still queue without bound. Being over the
+    // line already disqualifies it; the message is not sized individually.
+    st.observers.retain(|obs| {
+        obs.gate.queued_bytes() < OBSERVER_BUDGET && obs.tx.send(msg.clone()).is_ok()
+    });
 }
 
 fn fan_out_output(st: &mut PaneState, bytes: &[u8], gate: &OutputGate) {
@@ -622,16 +637,25 @@ impl DaemonPane {
         death.probe_exit_code({
             let child = child.clone();
             move || {
-                let deadline = std::time::Instant::now() + Duration::from_secs(2);
+                let deadline = std::time::Instant::now() + EXIT_CODE_PROBE_WINDOW;
                 loop {
-                    let status = child.lock().ok()?.try_wait().ok()?;
-                    if let Some(status) = status {
-                        return Some(status.exit_code() as i32);
+                    // try_lock, never lock: DaemonPane::drop holds this very
+                    // mutex across a blocking child.wait(), and this probe runs
+                    // on the reader thread on its way to announcing the death.
+                    // Blocking for the lock would park the Exited notification
+                    // behind a wait() with no bound of its own; the deadline
+                    // below is the only thing clients should ever wait on.
+                    if let Ok(mut child) = child.try_lock() {
+                        match child.try_wait() {
+                            Ok(Some(status)) => return Some(status.exit_code() as i32),
+                            Ok(None) => {}
+                            Err(_) => return None,
+                        }
                     }
                     if std::time::Instant::now() >= deadline {
                         return None;
                     }
-                    std::thread::sleep(Duration::from_millis(10));
+                    std::thread::sleep(EXIT_CODE_PROBE_INTERVAL);
                 }
             }
         });
@@ -1327,9 +1351,18 @@ fn attach_subscriber(st: &mut PaneState, subscriber: Sender<DaemonMsg>) -> u64 {
     st.subscriber_epoch
 }
 
-fn observe_subscriber(st: &mut PaneState, observer: Sender<DaemonMsg>, gate: Arc<OutputGate>) -> u64 {
+fn observe_subscriber(
+    st: &mut PaneState,
+    observer: Sender<DaemonMsg>,
+    gate: Arc<OutputGate>,
+) -> u64 {
     st.observer_seq += 1;
     replay_state(st, &observer);
+    // The replay just queued the whole ring into this channel. Charge it, or
+    // the first budget check would read zero while a full scrollback is already
+    // sitting there unread — an observer that never drains would be allowed a
+    // ring plus a full budget before anyone noticed.
+    gate.add(st.ring.len);
     st.observers.push(Observer {
         id: st.observer_seq,
         tx: observer,
@@ -3010,12 +3043,17 @@ mod tests {
 
         let (observer_tx, observer_rx) = mpsc::channel();
         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_eq!(
+            st.subscriber_epoch, epoch,
+            "observing must not bump the controller epoch"
+        );
         assert!(st.subscriber.is_some(), "the controller keeps its seat");
 
         assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Size(_))));
         assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Snapshot(b)) if b == b"screen"));
-        assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Cwd(p)) if p == PathBuf::from("/work")));
+        assert!(
+            matches!(observer_rx.try_recv(), Ok(DaemonMsg::Cwd(p)) if p == PathBuf::from("/work"))
+        );
         assert!(
             controller_rx.try_recv().is_err(),
             "an observer joining must be invisible to the controller"
@@ -3028,7 +3066,10 @@ mod tests {
         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");
+        assert!(
+            observer_rx.try_recv().is_err(),
+            "a departed observer hears nothing"
+        );
     }
 
     #[test]
@@ -3076,7 +3117,10 @@ mod tests {
         drop(observer_rx);
 
         notify(&mut st, DaemonMsg::Output(b"x".to_vec()));
-        assert!(st.observers.is_empty(), "a dead observer must not accumulate");
+        assert!(
+            st.observers.is_empty(),
+            "a dead observer must not accumulate"
+        );
     }
 
     #[test]
@@ -3162,7 +3206,10 @@ mod tests {
         let (dead_tx, dead_rx) = mpsc::channel();
         DeathReporter::new(move || dead_tx.send(()).unwrap())
             .report(&with_observer_only, &AtomicBool::new(false));
-        assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Exited { code: None })));
+        assert!(matches!(
+            observer_rx.try_recv(),
+            Ok(DaemonMsg::Exited { code: None })
+        ));
         assert!(
             dead_rx.try_recv().is_ok(),
             "read-only observers must not keep a dead pane in the registry"
@@ -3181,8 +3228,14 @@ mod tests {
         let (dead_tx, dead_rx) = mpsc::channel();
         DeathReporter::new(move || dead_tx.send(()).unwrap())
             .report(&with_both, &AtomicBool::new(false));
-        assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Exited { code: None })));
-        assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Exited { code: None })));
+        assert!(matches!(
+            controller_rx.try_recv(),
+            Ok(DaemonMsg::Exited { code: None })
+        ));
+        assert!(matches!(
+            observer_rx.try_recv(),
+            Ok(DaemonMsg::Exited { code: None })
+        ));
         assert!(
             dead_rx.try_recv().is_err(),
             "an attached death is still the detach path's to reclaim"

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

@@ -372,6 +372,11 @@ fn handle_conn(stream: Stream, registry: Arc<Registry>) -> anyhow::Result<()> {
             }
         },
 
+        // The size is accepted but deliberately not applied: an observer is
+        // read-only, and resizing the pty for it would reflow the grid under
+        // the controller. The replay tells the observer the pane's real size;
+        // the field stays on the wire so a future observer-side viewport does
+        // not need another message kind.
         ClientMsg::Observe { pane_id, size: _ } => match registry.get(pane_id) {
             Some(pane) => stream_observer(pane, read_stream, write_stream),
             None => {
@@ -663,8 +668,6 @@ 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,
@@ -676,18 +679,15 @@ fn run_stream(
 ) -> anyhow::Result<()> {
     use std::io::Read as _;
 
+    // Blocking reads: when this controller is displaced its writer's channel
+    // closes, and the writer shuts our read side down on its way out (see
+    // spawn_writer), so the read below returns instead of hanging.
     let writer = spawn_writer(rx, write_stream, pane.gate());
-    let _ = read_stream.set_read_timeout(Some(CONTROL_POLL_INTERVAL));
 
     let mut killed = false;
-    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,
@@ -700,14 +700,12 @@ fn run_stream(
             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);
@@ -731,20 +729,14 @@ fn run_stream(
         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;
-            }
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
             Err(_) => break,
         }
     }
 
+    // Asked after the loop, before detach clears the seat: whoever no longer
+    // holds the epoch was preempted rather than having hung up on their own.
+    let displaced = !pane.controls(epoch);
     let reclaimable = pane.detach(epoch);
     if displaced {
         let _ = read_stream.shutdown(std::net::Shutdown::Both);
@@ -809,6 +801,14 @@ fn spawn_writer(
                 }
                 let _ = write_stream.flush();
             }
+            // This half and the connection's read half are try_clone'd views of
+            // one socket, so shutting the read side down here unblocks whoever
+            // is parked in read() on the other handle. The writer's channel
+            // closing IS the handover signal — a new controller replaces the
+            // subscriber, the old Sender drops, rx.recv() fails, and we land
+            // here — so the displaced reader wakes at once and nobody has to
+            // poll for it.
+            let _ = write_stream.shutdown(std::net::Shutdown::Read);
         })
         .expect("spawn daemon writer thread")
 }
@@ -1132,5 +1132,40 @@ mod tests {
             writer.join().unwrap();
             drop(tx);
         }
+
+        #[test]
+        fn a_closed_writer_channel_wakes_the_controller_parked_in_read() {
+            let (client, server) = UnixStream::pair().unwrap();
+            let mut reader = server.try_clone().expect("clone the read half");
+            let (tx, rx) = mpsc::channel::<DaemonMsg>();
+            let writer = spawn_writer(rx, server, Arc::new(crate::daemon::pane::OutputGate::new()));
+
+            // Exactly where run_stream sits between frames.
+            let parked = thread::spawn(move || {
+                let mut buf = [0u8; 64];
+                std::io::Read::read(&mut reader, &mut buf).ok()
+            });
+
+            // A handover: the pane seats a new controller and drops this one's
+            // Sender. Nothing else happens — no poll, no timeout.
+            drop(tx);
+
+            let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+            while !parked.is_finished() && std::time::Instant::now() < deadline {
+                thread::sleep(std::time::Duration::from_millis(5));
+            }
+            assert!(
+                parked.is_finished(),
+                "a displaced controller must be woken by the writer's shutdown; \
+                 if this hangs, run_stream is back to polling for the handover"
+            );
+            assert_eq!(
+                parked.join().unwrap(),
+                Some(0),
+                "the woken read must report EOF, not a partial frame"
+            );
+            writer.join().unwrap();
+            drop(client);
+        }
     }
 }

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

@@ -53,9 +53,7 @@ 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]
-        }
+        Some(i) if name.is_char_boundary(i) && name[i..].eq_ignore_ascii_case(".exe") => &name[..i],
         _ => name,
     }
 }
@@ -75,7 +73,9 @@ fn is_reapable_daemon_name(name: &str) -> bool {
         .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))
+        || DAEMON_EXE_STEMS
+            .iter()
+            .any(|stem| exe_names_equal(stem, name))
 }
 
 pub fn ensure_running() -> anyhow::Result<()> {

crates/tty7-core/src/daemon/ssh/mod.rs

@@ -379,10 +379,16 @@ impl SshManager {
         let mut routes: Vec<_> = conns
             .iter()
             .map(|(key, slot)| {
-                let connected = slot
-                    .try_lock()
-                    .map(|weak| weak.upgrade().is_some_and(|conn| conn.is_alive()))
-                    .unwrap_or(false);
+                let connected = match slot.try_lock() {
+                    Ok(weak) => weak.upgrade().is_some_and(|conn| conn.is_alive()),
+                    // The slot is held by whoever is opening or using this link
+                    // right now. Busy is not down — and callers act on this:
+                    // the CLI's `-m <machine>` refuses to route over a link it
+                    // is told is down, so guessing false here fails a perfectly
+                    // live connection the moment it gets used. SshConnection::
+                    // is_alive resolves its own lock contention the same way.
+                    Err(_) => true,
+                };
                 crate::daemon::control::RouteInfo {
                     key: key.as_str().to_string(),
                     kind: "ssh".to_string(),
@@ -691,6 +697,36 @@ mod tests {
         }
     }
 
+    #[test]
+    fn a_busy_link_reads_as_connected_rather_than_down() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .build()
+            .expect("build test runtime");
+        let mgr = SshManager {
+            runtime,
+            conns: Mutex::new(HashMap::new()),
+            forwards: SshForwardRegistry::default(),
+            probes: Mutex::new(HashMap::new()),
+        };
+        let slot: ConnSlot = Arc::new(tokio::sync::Mutex::new(Weak::new()));
+        mgr.conns
+            .lock()
+            .unwrap()
+            .insert(ConnectionKey::from_spec(&base_spec()), slot.clone());
+
+        // Someone is mid-operation on this link: opening a channel, running the
+        // remote bootstrap probe, anything that holds the slot for a moment.
+        let _busy = slot.try_lock().expect("nobody else holds it in this test");
+
+        let routes = mgr.routes();
+        assert_eq!(routes.len(), 1, "a busy link must still be listed");
+        assert!(
+            routes[0].connected,
+            "a link whose slot is momentarily held is busy, not down — calling it \
+             down makes `tty7 -m <machine>` refuse to route over a live connection"
+        );
+    }
+
     #[test]
     fn connection_key_includes_jump_chain() {
         let mut with_jump = base_spec();

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

@@ -1403,6 +1403,11 @@ mod sock {
         host: SharedHost,
         services: Services,
     ) -> io::Result<PathBuf> {
+        // Anchor uptime here, not at the first Status request. The GUI can host
+        // the control listener too, and there `run_with` never runs — so
+        // without this the OnceLock is first set by whoever asks, and every
+        // answer reports a server that just started.
+        super::server_started();
         let path = control_socket_path()?;
         let listener = bind_control_socket(&path)?;
         std::thread::Builder::new()
@@ -1436,6 +1441,9 @@ mod wsock {
         host: SharedHost,
         services: Services,
     ) -> io::Result<PathBuf> {
+        // See the unix arm: uptime is anchored where the listener opens, so a
+        // GUI-hosted control server does not report itself as freshly started.
+        super::server_started();
         let path = control_endpoint_path()?;
         if let Ok(live) = transport::connect_endpoint(CONTROL_PORT_FILE) {
             drop(live);
@@ -1585,8 +1593,7 @@ mod aggregate_tests {
         };
         let client = client_with(services);
 
-        let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap()
-        else {
+        let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() else {
             panic!("AgentStates must answer with ReplyOk::AgentStates");
         };
         assert_eq!(states.len(), 1);
@@ -1600,8 +1607,7 @@ mod aggregate_tests {
     fn aggregates_still_answer_when_this_process_serves_no_panes() {
         let client = client_with(Services::none());
 
-        let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap()
-        else {
+        let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() else {
             panic!("AgentStates must answer with ReplyOk::AgentStates");
         };
         assert!(states.is_empty());

crates/tty7-server/tests/pane_send_input.rs

@@ -227,9 +227,12 @@ fn a_preempting_attach_closes_the_displaced_controller_and_drops_its_input() {
 
     let _ = first.input(b"echo tty7_stale_input\r");
 
-    first
-        .set_recv_timeout(Some(EOF_WITHIN))
-        .expect("bound the displaced reads");
+    // Best effort: the handover closes this connection immediately, and
+    // setsockopt on a socket whose peer is gone fails (EINVAL on macOS). Not a
+    // problem — the earlier STREAM_WITHIN bound is still in force, so the reads
+    // below stay bounded either way, and an already-closed connection is
+    // precisely what this test wants to see.
+    let _ = first.set_recv_timeout(Some(EOF_WITHIN));
     let deadline = Instant::now() + EOF_WITHIN + Duration::from_secs(5);
     let eof = loop {
         match first.recv() {