feat(core): pane exit codes, read-only observe, procs and routed pane clients

The daemon now waits its pane child and puts the real exit code on the
Exited frame (and replays it to late subscribers of a dead pane) — the
prerequisite for tty7 run's code passthrough. The client library gains
PaneClient::observe (read-only replay+stream), PaneClient::procs, and
PaneClient::routed for reaching a remote machine's pane daemon over the
local server's ROUTE frame, mirroring ControlClient::routed.

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

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

@@ -7,7 +7,9 @@ pub use pane::{PaneClient, PaneInput, PaneOutput, PaneSession};
 pub use crate::daemon::control::{
     ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ControlResponse, ReplyOk,
 };
-pub use crate::daemon::protocol::{DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize};
+pub use crate::daemon::protocol::{
+    DaemonMsg, DaemonVersion, PaneInfo, PaneProcs, ShellSpec, WinSize,
+};
 pub use crate::daemon::router::RouteTarget;
 
 #[cfg(test)]

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

@@ -3,9 +3,10 @@ use std::path::PathBuf;
 use std::time::Duration;
 
 use crate::daemon::protocol::{
-    ClientMsg, DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize, is_error_kind,
+    ClientMsg, DaemonMsg, DaemonVersion, PaneInfo, PaneProcs, ShellSpec, WinSize, is_error_kind,
     peek_frame_kind, take_frame,
 };
+use crate::daemon::router::{RouteAction, RouteChannel, RouteHeader, RouteTarget, negotiate};
 use crate::daemon::transport;
 
 const OPEN_REPLY_WAIT: Duration = Duration::from_secs(15);
@@ -15,6 +16,7 @@ enum PaneEndpoint {
     #[default]
     Local,
     At(PathBuf),
+    Routed(RouteTarget),
 }
 
 #[derive(Clone, Debug, Default)]
@@ -35,10 +37,27 @@ impl PaneClient {
         }
     }
 
+    pub fn routed(target: RouteTarget) -> PaneClient {
+        PaneClient {
+            endpoint: PaneEndpoint::Routed(target),
+        }
+    }
+
     fn open(&self) -> io::Result<transport::Stream> {
         match &self.endpoint {
             PaneEndpoint::Local => transport::connect(),
             PaneEndpoint::At(path) => transport::connect_endpoint_at(path),
+            PaneEndpoint::Routed(target) => {
+                let mut stream = transport::connect()?;
+                let header = RouteHeader {
+                    target: target.clone(),
+                    server_command: None,
+                    channel: RouteChannel::Pane,
+                    action: RouteAction::Forward,
+                };
+                negotiate(&mut stream, &header)?;
+                Ok(stream)
+            }
         }
     }
 
@@ -67,6 +86,16 @@ impl PaneClient {
         ClientMsg::Kill { pane_id }.encode(&mut stream)
     }
 
+    pub fn procs(&self, pane_id: u64) -> io::Result<PaneProcs> {
+        let mut stream = self.open()?;
+        ClientMsg::QueryProcs { pane_id }.encode(&mut stream)?;
+        match DaemonMsg::read(&mut stream)? {
+            DaemonMsg::Procs(procs) => Ok(procs),
+            DaemonMsg::Error(message) => Err(io::Error::other(message)),
+            other => Err(unexpected_reply("QueryProcs", &other)),
+        }
+    }
+
     pub fn spawn(
         &self,
         cwd: Option<PathBuf>,
@@ -81,6 +110,10 @@ impl PaneClient {
     pub fn attach(&self, pane_id: u64, size: WinSize) -> io::Result<PaneSession> {
         PaneSession::attach_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT)
     }
+
+    pub fn observe(&self, pane_id: u64, size: WinSize) -> io::Result<PaneSession> {
+        PaneSession::observe_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT)
+    }
 }
 
 #[derive(Debug)]
@@ -135,9 +168,28 @@ impl PaneSession {
         reply_wait: Duration,
     ) -> io::Result<PaneSession> {
         ClientMsg::Attach { pane_id, size }.encode(&mut stream)?;
+        PaneSession::checked(stream, "Attach", pane_id, reply_wait)
+    }
+
+    pub(crate) fn observe_over(
+        mut stream: transport::Stream,
+        pane_id: u64,
+        size: WinSize,
+        reply_wait: Duration,
+    ) -> io::Result<PaneSession> {
+        ClientMsg::Observe { pane_id, size }.encode(&mut stream)?;
+        PaneSession::checked(stream, "Observe", pane_id, reply_wait)
+    }
+
+    fn checked(
+        stream: transport::Stream,
+        request: &str,
+        pane_id: u64,
+        reply_wait: Duration,
+    ) -> io::Result<PaneSession> {
         let mut session = PaneSession::over(stream, pane_id)?;
         session.set_recv_timeout(Some(reply_wait))?;
-        let verdict = session.output.refusal_check(pane_id);
+        let verdict = session.output.refusal_check(request, pane_id);
         session.set_recv_timeout(None)?;
         verdict?;
         Ok(session)
@@ -252,14 +304,14 @@ impl PaneOutput {
         self.reader.set_read_timeout(wait)
     }
 
-    fn refusal_check(&mut self, pane_id: u64) -> io::Result<()> {
+    fn refusal_check(&mut self, request: &str, pane_id: u64) -> io::Result<()> {
         let mut scratch = [0u8; 4096];
         loop {
             if let Some(kind) = peek_frame_kind(&self.buffered) {
                 if !is_error_kind(kind) {
                     return Ok(());
                 }
-                return Err(self.refusal(pane_id));
+                return Err(self.refusal(request, pane_id));
             }
             match self.reader.read(&mut scratch) {
                 Ok(0) => {
@@ -267,7 +319,7 @@ impl PaneOutput {
                         io::ErrorKind::UnexpectedEof,
                         format!(
                             "the daemon closed the connection without answering \
-                             Attach for pane {pane_id}"
+                             {request} for pane {pane_id}"
                         ),
                     ));
                 }
@@ -279,13 +331,13 @@ impl PaneOutput {
         }
     }
 
-    fn refusal(&mut self, pane_id: u64) -> io::Error {
+    fn refusal(&mut self, request: &str, pane_id: u64) -> io::Error {
         match self.recv() {
             Ok(DaemonMsg::Error(message)) => {
-                io::Error::other(format!("daemon refused Attach: {message}"))
+                io::Error::other(format!("daemon refused {request}: {message}"))
             }
-            Ok(other) => unexpected_reply("Attach", &other),
-            Err(_) => io::Error::other(format!("daemon refused Attach for pane {pane_id}")),
+            Ok(other) => unexpected_reply(request, &other),
+            Err(_) => io::Error::other(format!("daemon refused {request} for pane {pane_id}")),
         }
     }
 }
@@ -413,6 +465,59 @@ mod tests {
         server.join().expect("server thread");
     }
 
+    #[test]
+    fn observe_opens_read_only_and_keeps_the_replay() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            match ClientMsg::read(&mut r).expect("read Observe") {
+                ClientMsg::Observe { pane_id, .. } => assert_eq!(pane_id, 9),
+                other => panic!("expected Observe, got {other:?}"),
+            }
+            DaemonMsg::Size(size()).encode(&mut w).expect("replay size");
+            DaemonMsg::Snapshot(b"ring contents".to_vec())
+                .encode(&mut w)
+                .expect("replay snapshot");
+        });
+
+        let mut session =
+            PaneSession::observe_over(client_end, 9, size(), REPLY_WAIT).expect("observe");
+        match session.recv().expect("replayed size") {
+            DaemonMsg::Size(_) => {}
+            other => panic!("expected Size, got {other:?}"),
+        }
+        match session.recv().expect("replayed snapshot") {
+            DaemonMsg::Snapshot(bytes) => assert_eq!(bytes, b"ring contents"),
+            other => panic!("expected Snapshot, got {other:?}"),
+        }
+        server.join().expect("server thread");
+    }
+
+    #[test]
+    fn an_observe_refusal_carries_the_daemons_message() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            match ClientMsg::read(&mut r).expect("read Observe") {
+                ClientMsg::Observe { pane_id, .. } => assert_eq!(pane_id, 42),
+                other => panic!("expected Observe, got {other:?}"),
+            }
+            DaemonMsg::Error("no such pane 42".into())
+                .encode(&mut w)
+                .expect("refuse the observe");
+        });
+
+        let err = PaneSession::observe_over(client_end, 42, size(), REPLY_WAIT)
+            .expect_err("observing a missing pane must fail");
+        assert!(
+            err.to_string().contains("no such pane 42"),
+            "the daemon's reason was lost: {err}"
+        );
+        server.join().expect("server thread");
+    }
+
     #[test]
     fn an_accepted_attach_keeps_the_replay_it_peeked_at() {
         let (client_end, server_end) = stream_pair();

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

@@ -429,6 +429,7 @@ struct PaneState {
     agent_argv: Option<Vec<String>>,
     agent_session: Option<crate::core::cli_agent::AgentSessionState>,
     alive: bool,
+    exit_code: Option<i32>,
 }
 
 fn notify(st: &mut PaneState, msg: DaemonMsg) {
@@ -451,7 +452,7 @@ struct ForegroundProbes {
 
 struct PtyBackend {
     master: Arc<Mutex<Box<dyn MasterPty + Send>>>,
-    child: Mutex<Box<dyn Child + Send + Sync>>,
+    child: Arc<Mutex<Box<dyn Child + Send + Sync>>>,
     #[cfg_attr(windows, allow(dead_code))]
     shell_pid: Option<u32>,
     integration_dir: Option<PathBuf>,
@@ -477,6 +478,7 @@ pub struct DaemonPane {
 struct DeathReporter {
     reported: AtomicBool,
     on_dead: Mutex<Option<Box<dyn FnOnce() + Send>>>,
+    exit_code: Mutex<Option<Box<dyn FnMut() -> Option<i32> + Send>>>,
 }
 
 impl DeathReporter {
@@ -484,15 +486,27 @@ impl DeathReporter {
         Self {
             reported: AtomicBool::new(false),
             on_dead: Mutex::new(Some(Box::new(on_dead))),
+            exit_code: Mutex::new(None),
         }
     }
 
+    fn probe_exit_code(&self, probe: impl FnMut() -> Option<i32> + Send + 'static) {
+        *self.exit_code.lock().unwrap() = Some(Box::new(probe));
+    }
+
     fn report(&self, state: &Mutex<PaneState>, shutting_down: &AtomicBool) {
         if self.reported.swap(true, Ordering::SeqCst) {
             return;
         }
+        let code = self
+            .exit_code
+            .lock()
+            .unwrap()
+            .as_mut()
+            .and_then(|probe| probe());
         let mut st = state.lock().unwrap();
         st.alive = false;
+        st.exit_code = code;
         let pane = st.id;
         if shutting_down.load(Ordering::SeqCst) {
             drop(st);
@@ -500,7 +514,7 @@ impl DeathReporter {
             return;
         }
         let subscribed = st.subscriber.is_some();
-        notify(&mut st, DaemonMsg::Exited { code: None });
+        notify(&mut st, DaemonMsg::Exited { code });
         drop(st);
         crate::core::machine::observe_pane(pane, |p| p.live = false);
         if subscribed {
@@ -529,6 +543,7 @@ impl DaemonPane {
 
         let child = pair.slave.spawn_command(spawn.cmd)?;
         let shell_pid = child.process_id();
+        let child = Arc::new(Mutex::new(child));
 
         drop(pair.slave);
 
@@ -549,6 +564,7 @@ impl DaemonPane {
             agent_session: None,
             agent_argv: None,
             alive: true,
+            exit_code: None,
         }));
         let shutting_down = Arc::new(AtomicBool::new(false));
         let gate = Arc::new(OutputGate::new());
@@ -560,7 +576,7 @@ impl DaemonPane {
             owner,
             backend: PaneBackend::Pty(PtyBackend {
                 master: master.clone(),
-                child: Mutex::new(child),
+                child: child.clone(),
                 shell_pid,
                 integration_dir: spawn.integration_dir,
             }),
@@ -573,6 +589,22 @@ impl DaemonPane {
         });
 
         let death = Arc::new(DeathReporter::new(on_dead));
+        death.probe_exit_code({
+            let child = child.clone();
+            move || {
+                let deadline = std::time::Instant::now() + Duration::from_secs(2);
+                loop {
+                    let status = child.lock().ok()?.try_wait().ok()?;
+                    if let Some(status) = status {
+                        return Some(status.exit_code() as i32);
+                    }
+                    if std::time::Instant::now() >= deadline {
+                        return None;
+                    }
+                    std::thread::sleep(Duration::from_millis(10));
+                }
+            }
+        });
 
         #[cfg(windows)]
         Self::spawn_exit_monitor(
@@ -639,6 +671,7 @@ impl DaemonPane {
             agent_session: None,
             agent_argv: None,
             alive: true,
+            exit_code: None,
         }));
         let shutting_down = Arc::new(AtomicBool::new(false));
         let gate = Arc::new(OutputGate::new());
@@ -1257,7 +1290,7 @@ fn replay_state(st: &PaneState, subscriber: &Sender<DaemonMsg>) {
         let _ = subscriber.send(DaemonMsg::AgentStatus(st.agent_session.clone()));
     }
     if !st.alive {
-        let _ = subscriber.send(DaemonMsg::Exited { code: None });
+        let _ = subscriber.send(DaemonMsg::Exited { code: st.exit_code });
     }
 }
 
@@ -2591,6 +2624,7 @@ mod tests {
             agent_session: None,
             agent_argv: None,
             alive,
+            exit_code: None,
         }
     }
 

crates/tty7-server/tests/client_lib.rs

@@ -286,6 +286,34 @@ fn a_spawned_pane_streams_its_output_and_its_exit() {
     );
 }
 
+#[test]
+fn a_one_shot_pane_reports_its_real_exit_code() {
+    let daemon = Daemon::start();
+    let mut session = daemon
+        .panes()
+        .spawn(
+            None,
+            size(),
+            Some(one_shot_shell("exit 5")),
+            Some("client-lib-test".into()),
+            None,
+        )
+        .expect("spawn a failing one-shot pane");
+    session
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+    loop {
+        match session.recv() {
+            Ok(DaemonMsg::Exited { code }) => {
+                assert_eq!(code, Some(5), "the child's code must ride the Exited frame");
+                break;
+            }
+            Ok(_) => {}
+            Err(e) => panic!("pane stream ended before Exited: {e}"),
+        }
+    }
+}
+
 #[test]
 fn input_reaches_the_shell_and_a_reattach_replays_it() {
     let daemon = Daemon::start();