diff --git a/crates/tty7-core/src/client/mod.rs b/crates/tty7-core/src/client/mod.rs index e2e654bb..8e938ad6 100644 --- a/crates/tty7-core/src/client/mod.rs +++ b/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)] diff --git a/crates/tty7-core/src/client/pane.rs b/crates/tty7-core/src/client/pane.rs index 1bb98b99..251514f3 100644 --- a/crates/tty7-core/src/client/pane.rs +++ b/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 { 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 { + 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, @@ -81,6 +110,10 @@ impl PaneClient { pub fn attach(&self, pane_id: u64, size: WinSize) -> io::Result { PaneSession::attach_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT) } + + pub fn observe(&self, pane_id: u64, size: WinSize) -> io::Result { + PaneSession::observe_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT) + } } #[derive(Debug)] @@ -135,9 +168,28 @@ impl PaneSession { reply_wait: Duration, ) -> io::Result { 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 { + 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 { 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(); diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 7b929ab5..4e0924fc 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -429,6 +429,7 @@ struct PaneState { agent_argv: Option>, agent_session: Option, alive: bool, + exit_code: Option, } fn notify(st: &mut PaneState, msg: DaemonMsg) { @@ -451,7 +452,7 @@ struct ForegroundProbes { struct PtyBackend { master: Arc>>, - child: Mutex>, + child: Arc>>, #[cfg_attr(windows, allow(dead_code))] shell_pid: Option, integration_dir: Option, @@ -477,6 +478,7 @@ pub struct DaemonPane { struct DeathReporter { reported: AtomicBool, on_dead: Mutex>>, + exit_code: Mutex Option + 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 + Send + 'static) { + *self.exit_code.lock().unwrap() = Some(Box::new(probe)); + } + fn report(&self, state: &Mutex, 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) { 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, } } diff --git a/crates/tty7-server/tests/client_lib.rs b/crates/tty7-server/tests/client_lib.rs index 47353cd5..0cbe80bb 100644 --- a/crates/tty7-server/tests/client_lib.rs +++ b/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();