From 5511818af9292e469e241f265bbb404ea816cd01 Mon Sep 17 00:00:00 2001 From: thomas Date: Fri, 31 Jul 2026 10:18:16 +0800 Subject: [PATCH 1/2] feat(core): pane exit codes, read-only observe, procs and routed pane clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv --- crates/tty7-core/src/client/mod.rs | 4 +- crates/tty7-core/src/client/pane.rs | 123 +++++++++++++++++++++++-- crates/tty7-core/src/daemon/pane.rs | 42 ++++++++- crates/tty7-server/tests/client_lib.rs | 28 ++++++ 4 files changed, 183 insertions(+), 14 deletions(-) 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(); From fafb0019f4e6b0bada766ae59184c673e0c0f2ad Mon Sep 17 00:00:00 2001 From: thomas Date: Fri, 31 Jul 2026 10:18:30 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat(cli):=20real=20server=20backend=20?= =?UTF-8?q?=E2=80=94=20every=20non-interactive=20verb=20goes=20live?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StubBackend is gone; RealBackend lazily opens a ControlClient for control ops and a PaneClient for pane ops. Backend::spawn_shell now returns the daemon-assigned pane id, so every creating verb (new, tab new, split) spawns first and seeds the tree op with the real id — client-side pane-id allocation is deleted. Live end to end: ls, ws/tab/pane verbs, new, send (attach-input-detach), capture (observer replay; --scrollback = whole ring, default = the ring's last segment), procs, run (streams output, passes the child's exit code through, --keep leaves the pane), events (human lines or NDJSON), agents, status, machine ls, and doctor's server half (reachability, dialect, status, links). tty7 server start|stop| restart|logs manage a sibling/PATH/TTY7_SERVER_EXE tty7-server. -m routes both channels over the local server's link, resolved against Routes by key or bare host; jump/proxy-chained keys are refused with the reason. Interactive attach stays stubbed for the next slice. A harness-free e2e suite drives the compiled tty7.exe against an isolated real server. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv --- Cargo.lock | 2 + crates/tty7-cli/Cargo.toml | 12 + crates/tty7-cli/src/backend.rs | 117 +++++---- crates/tty7-cli/src/backend/real.rs | 364 ++++++++++++++++++++++++++++ crates/tty7-cli/src/cli.rs | 7 + crates/tty7-cli/src/commands.rs | 258 +++++++++++++++----- crates/tty7-cli/src/main.rs | 3 +- crates/tty7-cli/src/output.rs | 55 +++++ crates/tty7-cli/src/resolve.rs | 18 -- crates/tty7-cli/src/server.rs | 177 ++++++++++++++ crates/tty7-cli/tests/cli_e2e.rs | 290 ++++++++++++++++++++++ 11 files changed, 1165 insertions(+), 138 deletions(-) create mode 100644 crates/tty7-cli/src/backend/real.rs create mode 100644 crates/tty7-cli/src/server.rs create mode 100644 crates/tty7-cli/tests/cli_e2e.rs diff --git a/Cargo.lock b/Cargo.lock index 430f1cdc..4c003231 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9694,7 +9694,9 @@ version = "26.7.6" dependencies = [ "anyhow", "clap", + "libc", "serde_json", + "tempfile", "tty7-core", ] diff --git a/crates/tty7-cli/Cargo.toml b/crates/tty7-cli/Cargo.toml index 9b0b7da3..38bdc543 100644 --- a/crates/tty7-cli/Cargo.toml +++ b/crates/tty7-cli/Cargo.toml @@ -29,5 +29,17 @@ serde_json.workspace = true # user, same as the GUI package's own single-user pins (regex, memchr, …). clap = { version = "4", features = ["derive"] } +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[dev-dependencies] +tempfile = "3" + +# The end-to-end suite drives the compiled tty7.exe against a real isolated +# server; its own main doubles as that server process, so no libtest harness. +[[test]] +name = "cli_e2e" +harness = false + [lints] workspace = true diff --git a/crates/tty7-cli/src/backend.rs b/crates/tty7-cli/src/backend.rs index e6cbfa39..20938a22 100644 --- a/crates/tty7-cli/src/backend.rs +++ b/crates/tty7-cli/src/backend.rs @@ -1,8 +1,12 @@ -use anyhow::{Result, bail}; +use anyhow::Result; use tty7_core::core::session::WorkspaceId; -use tty7_core::daemon::control::{ControlEvent, ControlRequest, ReplyOk}; +use tty7_core::daemon::control::{ControlEvent, ControlHelloOk, ControlRequest, ReplyOk}; use tty7_core::daemon::protocol::PaneProcs; +mod real; + +pub use real::RealBackend; + #[derive(Debug, Clone, PartialEq)] pub struct RunSpec { pub workspace: Option, @@ -14,8 +18,9 @@ pub struct RunSpec { pub trait Backend { fn control(&mut self, req: ControlRequest) -> Result; - fn spawn_shell(&mut self, pane: u64, workspace: WorkspaceId, cwd: Option) - -> Result<()>; + fn hello(&mut self) -> Result; + + fn spawn_shell(&mut self, workspace: WorkspaceId, cwd: Option) -> Result; fn send_input(&mut self, pane: u64, bytes: Vec) -> Result<()>; @@ -30,50 +35,6 @@ pub trait Backend { fn events(&mut self, on_event: &mut dyn FnMut(ControlEvent) -> Result<()>) -> Result<()>; } -pub struct StubBackend; - -const NOT_WIRED: &str = - "the tty7 CLI is not wired to a server yet — the transport client lands in the next slice"; - -impl Backend for StubBackend { - fn control(&mut self, _req: ControlRequest) -> Result { - bail!(NOT_WIRED) - } - - fn spawn_shell( - &mut self, - _pane: u64, - _workspace: WorkspaceId, - _cwd: Option, - ) -> Result<()> { - bail!(NOT_WIRED) - } - - fn send_input(&mut self, _pane: u64, _bytes: Vec) -> Result<()> { - bail!(NOT_WIRED) - } - - fn capture(&mut self, _pane: u64, _scrollback: bool) -> Result { - bail!(NOT_WIRED) - } - - fn procs(&mut self, _pane: u64) -> Result { - bail!(NOT_WIRED) - } - - fn attach_pane(&mut self, _pane: u64) -> Result<()> { - bail!(NOT_WIRED) - } - - fn run(&mut self, _spec: RunSpec) -> Result { - bail!(NOT_WIRED) - } - - fn events(&mut self, _on_event: &mut dyn FnMut(ControlEvent) -> Result<()>) -> Result<()> { - bail!(NOT_WIRED) - } -} - #[cfg(test)] pub mod mock { use std::collections::VecDeque; @@ -81,23 +42,45 @@ pub mod mock { use anyhow::{Result, anyhow}; use tty7_core::core::machine::Machine; use tty7_core::core::session::WorkspaceId; - use tty7_core::daemon::control::{ControlEvent, ControlRequest, ReplyOk}; - use tty7_core::daemon::protocol::PaneProcs; + use tty7_core::daemon::control::{ + CONTROL_VERSION, ControlEvent, ControlHelloOk, ControlRequest, ReplyOk, feature, + }; + use tty7_core::daemon::protocol::{PROTOCOL_VERSION, PaneProcs}; use super::{Backend, RunSpec}; - #[derive(Default)] pub struct MockBackend { pub machine: Machine, pub replies: VecDeque, pub control_calls: Vec, - pub spawned: Vec<(u64, WorkspaceId, Option)>, + pub spawned: Vec<(WorkspaceId, Option)>, + pub next_spawn_id: u64, pub sent: Vec<(u64, Vec)>, pub captured: Vec<(u64, bool)>, pub capture_text: String, pub procs_calls: Vec, pub procs_reply: PaneProcs, pub runs: Vec, + pub events: Vec, + } + + impl Default for MockBackend { + fn default() -> MockBackend { + MockBackend { + machine: Machine::default(), + replies: VecDeque::new(), + control_calls: Vec::new(), + spawned: Vec::new(), + next_spawn_id: 6, + sent: Vec::new(), + captured: Vec::new(), + capture_text: String::new(), + procs_calls: Vec::new(), + procs_reply: PaneProcs::default(), + runs: Vec::new(), + events: Vec::new(), + } + } } impl MockBackend { @@ -119,14 +102,23 @@ pub mod mock { Ok(self.replies.pop_front().unwrap_or(ReplyOk::Unit)) } - fn spawn_shell( - &mut self, - pane: u64, - workspace: WorkspaceId, - cwd: Option, - ) -> Result<()> { - self.spawned.push((pane, workspace, cwd)); - Ok(()) + fn hello(&mut self) -> Result { + Ok(ControlHelloOk { + control_version: CONTROL_VERSION, + protocol_version: PROTOCOL_VERSION, + build: "mock".into(), + separator: '\\', + home: "C:\\Users\\mock".into(), + features: vec![feature::CONTROL.into(), feature::MACHINE_TREE.into()], + instance: "mock-instance".into(), + }) + } + + fn spawn_shell(&mut self, workspace: WorkspaceId, cwd: Option) -> Result { + self.spawned.push((workspace, cwd)); + let id = self.next_spawn_id; + self.next_spawn_id += 1; + Ok(id) } fn send_input(&mut self, pane: u64, bytes: Vec) -> Result<()> { @@ -153,7 +145,10 @@ pub mod mock { Ok(0) } - fn events(&mut self, _on_event: &mut dyn FnMut(ControlEvent) -> Result<()>) -> Result<()> { + fn events(&mut self, on_event: &mut dyn FnMut(ControlEvent) -> Result<()>) -> Result<()> { + for event in self.events.drain(..) { + on_event(event)?; + } Ok(()) } } diff --git a/crates/tty7-cli/src/backend/real.rs b/crates/tty7-cli/src/backend/real.rs new file mode 100644 index 00000000..d77fcd53 --- /dev/null +++ b/crates/tty7-cli/src/backend/real.rs @@ -0,0 +1,364 @@ +use std::io::Write as _; +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{Context as _, Result, anyhow, bail}; +use serde_json::json; +use tty7_core::client::{ControlClient, PaneClient}; +use tty7_core::core::session::WorkspaceId; +use tty7_core::daemon::control::{ + ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ReplyOk, RouteInfo, +}; +use tty7_core::daemon::protocol::{DaemonMsg, PaneProcs, ShellSpec, WinSize}; +use tty7_core::daemon::router::RouteTarget; + +use super::{Backend, RunSpec}; + +const SESSION_SIZE: WinSize = WinSize { + cols: 120, + rows: 30, + cell_w: 8, + cell_h: 16, +}; + +const REPLAY_FIRST_WAIT: Duration = Duration::from_secs(10); +const REPLAY_SETTLE: Duration = Duration::from_millis(300); + +const NOT_RUNNING: &str = + "could not reach the tty7 server on this machine — `tty7 server start` brings one up"; + +pub struct RealBackend { + machine: Option, + route: Option, + control: Option, + panes: Option, +} + +impl RealBackend { + pub fn new(machine: Option) -> RealBackend { + RealBackend { + machine, + route: None, + control: None, + panes: None, + } + } + + fn hello_msg() -> ControlHello { + ControlHello::host_rpc(format!("tty7-cli-{}", std::process::id()), hostname()) + } + + fn route(&mut self) -> Result> { + let Some(name) = self.machine.clone() else { + return Ok(None); + }; + if let Some(target) = &self.route { + return Ok(Some(target.clone())); + } + let local = ControlClient::connect(&Self::hello_msg()).context(NOT_RUNNING)?; + let routes = match local + .request(ControlRequest::Routes) + .context("asking the local server for its machine links")? + { + ReplyOk::Routes(routes) => routes, + other => bail!("the server answered Routes with {other:?}"), + }; + local.close(); + let target = resolve_route(&name, &routes)?; + self.route = Some(target.clone()); + Ok(Some(target)) + } + + fn control_client(&mut self) -> Result<&ControlClient> { + if self.control.is_none() { + let hello = Self::hello_msg(); + let client = match self.route()? { + Some(target) => ControlClient::routed(target, &hello).with_context(|| { + format!( + "routing to machine '{}' through the local server", + self.machine.as_deref().unwrap_or_default() + ) + })?, + None => ControlClient::connect(&hello).context(NOT_RUNNING)?, + }; + self.control = Some(client); + } + Ok(self.control.as_ref().expect("just filled")) + } + + fn pane_client(&mut self) -> Result<&PaneClient> { + if self.panes.is_none() { + let client = match self.route()? { + Some(target) => PaneClient::routed(target), + None => PaneClient::local(), + }; + self.panes = Some(client); + } + Ok(self.panes.as_ref().expect("just filled")) + } +} + +impl Backend for RealBackend { + fn control(&mut self, req: ControlRequest) -> Result { + let reply = self.control_client()?.request(req)?; + Ok(reply) + } + + fn hello(&mut self) -> Result { + Ok(self.control_client()?.hello().clone()) + } + + fn spawn_shell(&mut self, workspace: WorkspaceId, cwd: Option) -> Result { + let session = self + .pane_client()? + .spawn( + cwd.map(PathBuf::from), + SESSION_SIZE, + None, + Some("tty7-cli".into()), + Some(workspace.to_string()), + ) + .context("spawning a shell")?; + let pane = session.pane_id(); + session.detach()?; + Ok(pane) + } + + fn send_input(&mut self, pane: u64, bytes: Vec) -> Result<()> { + let mut session = self + .pane_client()? + .attach(pane, SESSION_SIZE) + .with_context(|| format!("attaching to pane %{pane}"))?; + session.input(&bytes)?; + session.detach()?; + Ok(()) + } + + fn capture(&mut self, pane: u64, scrollback: bool) -> Result { + let mut session = self + .pane_client()? + .observe(pane, SESSION_SIZE) + .with_context(|| format!("observing pane %{pane}"))?; + session.set_recv_timeout(Some(REPLAY_FIRST_WAIT))?; + let mut snapshots: Vec> = Vec::new(); + loop { + match session.recv() { + Ok(DaemonMsg::Snapshot(bytes)) => { + snapshots.push(bytes); + session.set_recv_timeout(Some(REPLAY_SETTLE))?; + } + Ok(DaemonMsg::Output(_)) | Ok(DaemonMsg::Exited { .. }) => break, + Ok(_) => session.set_recv_timeout(Some(REPLAY_SETTLE))?, + Err(e) if timed_out(&e) => break, + Err(e) => return Err(anyhow!(e).context("reading the pane replay")), + } + } + let _ = session.detach(); + let bytes: Vec = if scrollback { + snapshots.concat() + } else { + snapshots.pop().unwrap_or_default() + }; + Ok(String::from_utf8_lossy(&bytes).into_owned()) + } + + fn procs(&mut self, pane: u64) -> Result { + let procs = self.pane_client()?.procs(pane)?; + Ok(procs) + } + + fn attach_pane(&mut self, _pane: u64) -> Result<()> { + bail!("interactive `tty7 attach %pane` is not wired yet — it lands in the next slice") + } + + fn run(&mut self, spec: RunSpec) -> Result { + let (program, args) = spec + .command + .split_first() + .ok_or_else(|| anyhow!("run needs a command after `--`"))?; + let shell = ShellSpec { + program: program.clone(), + args: args.to_vec(), + args_are_tty7_defaults: false, + }; + let mut session = self + .pane_client()? + .spawn( + spec.cwd.map(PathBuf::from), + SESSION_SIZE, + Some(shell), + Some("tty7-cli".into()), + spec.workspace.map(|ws| ws.to_string()), + ) + .with_context(|| format!("spawning `{program}`"))?; + let mut stdout = std::io::stdout().lock(); + let code = loop { + match session.recv() { + Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => { + stdout.write_all(&bytes)?; + stdout.flush()?; + } + Ok(DaemonMsg::Exited { code }) => break code.unwrap_or(1), + Ok(_) => {} + Err(e) => return Err(anyhow!(e).context("streaming the command's output")), + } + }; + if spec.keep { + session.detach()?; + } else { + session.kill()?; + } + Ok(code) + } + + fn events(&mut self, on_event: &mut dyn FnMut(ControlEvent) -> Result<()>) -> Result<()> { + let client = self.control_client()?; + for event in client.events() { + on_event(event)?; + } + Ok(()) + } +} + +fn timed_out(e: &std::io::Error) -> bool { + matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) +} + +fn hostname() -> String { + std::env::var("COMPUTERNAME") + .or_else(|_| std::env::var("HOSTNAME")) + .unwrap_or_else(|_| "tty7-cli".to_string()) +} + +fn resolve_route(name: &str, routes: &[RouteInfo]) -> Result { + let matches: Vec<&RouteInfo> = routes.iter().filter(|r| route_matches(name, r)).collect(); + match matches.as_slice() { + [one] => target_for(one), + [] if routes.is_empty() => bail!( + "the local server holds no machine links — connect one from the GUI first \ + (`tty7 machine ls` shows them)" + ), + [] => bail!( + "no machine '{name}' — known machines: {}", + keys(routes.iter()) + ), + many => bail!( + "'{name}' names {} machines — use the full key: {}", + many.len(), + keys(many.iter().copied()) + ), + } +} + +fn keys<'a>(routes: impl Iterator) -> String { + routes + .map(|r| r.key.as_str()) + .collect::>() + .join(", ") +} + +fn route_matches(name: &str, route: &RouteInfo) -> bool { + route.key == name || host_of(&route.key) == Some(name) +} + +fn host_of(key: &str) -> Option<&str> { + let first = key.split('|').next()?; + let after_user = first.split('@').nth(1)?; + after_user.split(':').next() +} + +fn target_for(route: &RouteInfo) -> Result { + if route.kind != "ssh" { + bail!( + "machine '{}' is a {} link — the CLI can only route over ssh links yet", + route.key, + route.kind + ); + } + if route.key.contains('|') { + bail!( + "machine '{}' is reached through a jump/proxy chain, which the CLI cannot \ + rebuild from the link key yet — use the GUI for this machine", + route.key + ); + } + let (user, rest) = route + .key + .split_once('@') + .ok_or_else(|| anyhow!("unrecognized machine key '{}'", route.key))?; + let (host, port) = rest + .rsplit_once(':') + .ok_or_else(|| anyhow!("unrecognized machine key '{}'", route.key))?; + let port: u16 = port + .parse() + .map_err(|_| anyhow!("unrecognized machine key '{}'", route.key))?; + let spec = serde_json::from_value(json!({ + "user": user, + "host": host, + "port": port, + "auth_mode": "auto", + }))?; + Ok(RouteTarget::Ssh(Box::new(spec))) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn route(key: &str, kind: &str, connected: bool) -> RouteInfo { + RouteInfo { + key: key.into(), + kind: kind.into(), + connected, + } + } + + #[test] + fn a_machine_resolves_by_full_key_or_bare_host() { + let routes = vec![ + route("me@build-box:22", "ssh", true), + route("me@web-box:2222", "ssh", false), + ]; + for name in ["me@build-box:22", "build-box"] { + let RouteTarget::Ssh(spec) = resolve_route(name, &routes).unwrap() else { + panic!("ssh routes resolve to ssh targets"); + }; + assert_eq!(spec.user, "me"); + assert_eq!(spec.host, "build-box"); + assert_eq!(spec.port, 22); + } + let RouteTarget::Ssh(spec) = resolve_route("web-box", &routes).unwrap() else { + panic!("ssh routes resolve to ssh targets"); + }; + assert_eq!(spec.port, 2222); + } + + #[test] + fn unknown_and_ambiguous_names_list_the_candidates() { + let routes = vec![ + route("a@shared:22", "ssh", true), + route("b@shared:22", "ssh", true), + ]; + let err = resolve_route("nowhere", &routes).unwrap_err().to_string(); + assert!(err.contains("a@shared:22"), "{err}"); + assert!(err.contains("b@shared:22"), "{err}"); + + let err = resolve_route("shared", &routes).unwrap_err().to_string(); + assert!(err.contains("2 machines"), "{err}"); + + let err = resolve_route("anything", &[]).unwrap_err().to_string(); + assert!(err.contains("no machine links"), "{err}"); + } + + #[test] + fn chained_keys_are_refused_with_the_reason() { + let routes = vec![route("me@inner:22|jump:me@bastion:22", "ssh", true)]; + let err = resolve_route("me@inner:22|jump:me@bastion:22", &routes) + .unwrap_err() + .to_string(); + assert!(err.contains("jump/proxy chain"), "{err}"); + } +} diff --git a/crates/tty7-cli/src/cli.rs b/crates/tty7-cli/src/cli.rs index 6a92311f..426f2118 100644 --- a/crates/tty7-cli/src/cli.rs +++ b/crates/tty7-cli/src/cli.rs @@ -105,6 +105,13 @@ pub struct RunArgs { #[arg(long, value_name = "DIR", help = "Working directory for the command")] pub cwd: Option, + #[arg( + long, + value_name = "WORKSPACE", + help = "Workspace the pane belongs to; defaults to $TTY7_WS inside a tty7 shell" + )] + pub ws: Option, + #[arg( last = true, required = true, diff --git a/crates/tty7-cli/src/commands.rs b/crates/tty7-cli/src/commands.rs index 3c45c95e..b7ff044d 100644 --- a/crates/tty7-cli/src/commands.rs +++ b/crates/tty7-cli/src/commands.rs @@ -2,7 +2,10 @@ use anyhow::{Result, bail}; use serde_json::{Value, json}; use tty7_core::core::machine::{Axis, Machine, PaneSeed, Workspace}; use tty7_core::core::session::WorkspaceId; -use tty7_core::daemon::control::{ControlRequest, ReplyOk}; +use tty7_core::daemon::control::{ + CONTROL_VERSION, ControlEvent, ControlRequest, ReplyOk, +}; +use tty7_core::daemon::protocol::PROTOCOL_VERSION; use crate::address::{self, Address, Context, WorkspaceAddress}; use crate::backend::{Backend, RunSpec}; @@ -66,28 +69,19 @@ pub fn execute(cli: Cli, ctx: &Context, backend: &mut dyn Backend) -> Result pane_ls(ws.as_deref(), backend), Some(Command::Pane(PaneCmd::Close { target })) => pane_close(target.as_deref(), ctx, backend), Some(Command::Events) => events(json_mode, backend), - Some(Command::Agents) => bail!( - "`tty7 agents` needs ControlRequest::AgentStates, which this build's \ - control dialect does not have yet" - ), - Some(Command::Status) | Some(Command::Server(ServerCmd::Status)) => bail!( - "`tty7 status` needs ControlRequest::Status, which this build's \ - control dialect does not have yet" - ), - Some(Command::Machine(MachineCmd::Ls)) => bail!( - "`tty7 machine ls` needs ControlRequest::Routes, which this build's \ - control dialect does not have yet" - ), + Some(Command::Agents) => agents(backend), + Some(Command::Status) | Some(Command::Server(ServerCmd::Status)) => status(backend), + Some(Command::Machine(MachineCmd::Ls)) => machine_ls(backend), Some(Command::Machine(MachineCmd::Connect { .. })) | Some(Command::Machine(MachineCmd::Disconnect { .. })) => bail!( "managing machine links from the CLI is not implemented yet — \ use the GUI's connection manager for now" ), - Some(Command::Server(_)) => bail!( - "managing the local server process from the CLI is not implemented yet — \ - it arrives once the GUI stops bundling the server role" - ), - Some(Command::Doctor) => doctor(ctx), + Some(Command::Server(ServerCmd::Start)) => crate::server::start(), + Some(Command::Server(ServerCmd::Stop)) => crate::server::stop(), + Some(Command::Server(ServerCmd::Restart)) => crate::server::restart(), + Some(Command::Server(ServerCmd::Logs)) => crate::server::logs(), + Some(Command::Doctor) => doctor(ctx, backend), } } @@ -201,8 +195,6 @@ fn ws_detach(ws: &str, backend: &mut dyn Backend) -> Result { } fn new_workspace(path: Option, backend: &mut dyn Backend) -> Result { - let machine = fetch_machine(backend)?; - let pane = resolve::next_pane_id(&machine); let ws = match backend.control(ControlRequest::WorkspaceCreate { name: None, workspace: None, @@ -210,18 +202,18 @@ fn new_workspace(path: Option, backend: &mut dyn Backend) -> Result *ws, other => bail!("the server answered WorkspaceCreate with {other:?}"), }; + let pane = backend.spawn_shell(ws.id, path.clone())?; backend.control(ControlRequest::TabCreate { workspace: ws.id, at: None, pane: PaneSeed { pane, - cwd: path.clone(), + cwd: path, ssh_spec: None, agent: None, }, tab: None, })?; - backend.spawn_shell(pane, ws.id, path)?; report( ws.id.to_string(), json!({ "id": ws.id.to_string(), "pane": pane }), @@ -240,10 +232,13 @@ fn attach(target: &str, backend: &mut dyn Backend) -> Result { } fn run(args: RunArgs, ctx: &Context, backend: &mut dyn Backend) -> Result { - let workspace = ctx - .ws - .as_deref() - .and_then(|v| v.parse::().ok()); + let workspace = match args.ws.as_deref() { + Some(explicit) => { + let machine = fetch_machine(backend)?; + Some(resolve::workspace(&machine, &address::parse_workspace(explicit))?.id) + } + None => ctx.ws.as_deref().and_then(|v| v.parse::().ok()), + }; let code = backend.run(RunSpec { workspace, cwd: args.cwd, @@ -257,7 +252,6 @@ fn pane_split(args: SplitArgs, ctx: &Context, backend: &mut dyn Backend) -> Resu let pane = address::pane_or_context(args.target.as_deref(), ctx)?; let machine = fetch_machine(backend)?; let workspace = resolve::workspace_of_pane(&machine, pane)?.id; - let new = resolve::next_pane_id(&machine); let cwd = machine .panes .iter() @@ -268,6 +262,7 @@ fn pane_split(args: SplitArgs, ctx: &Context, backend: &mut dyn Backend) -> Resu } else { Axis::Vertical }; + let new = backend.spawn_shell(workspace, cwd.clone())?; backend.control(ControlRequest::PaneSplit { workspace, pane, @@ -275,13 +270,12 @@ fn pane_split(args: SplitArgs, ctx: &Context, backend: &mut dyn Backend) -> Resu ratio: args.ratio, new: PaneSeed { pane: new, - cwd: cwd.clone(), + cwd, ssh_spec: None, agent: None, }, first: false, })?; - backend.spawn_shell(new, workspace, cwd)?; report(format!("%{new}"), json!({ "pane": new })) } @@ -361,13 +355,13 @@ fn tab_new( ) -> Result { let machine = fetch_machine(backend)?; let id = resolve_ws(explicit, ctx, &machine)?; - let pane = resolve::next_pane_id(&machine); + let pane = backend.spawn_shell(id, cwd.clone())?; let tab = match backend.control(ControlRequest::TabCreate { workspace: id, at: None, pane: PaneSeed { pane, - cwd: cwd.clone(), + cwd, ssh_spec: None, agent: None, }, @@ -376,7 +370,6 @@ fn tab_new( ReplyOk::TabTree(tab) => *tab, other => bail!("the server answered TabCreate with {other:?}"), }; - backend.spawn_shell(pane, id, cwd)?; report( format!("%{pane}"), json!({ "tab": tab.id.to_string(), "pane": pane }), @@ -455,31 +448,126 @@ fn events(json_mode: bool, backend: &mut dyn Backend) -> Result { if json_mode { println!("{}", serde_json::to_string(&event)?); } else { - println!("{event:?}"); + println!("{}", event_line(&event)); } Ok(()) })?; report("", Value::Null) } -fn doctor(ctx: &Context) -> Result { +fn event_line(event: &ControlEvent) -> String { + match event { + ControlEvent::PaneExited { pane_id, code } => match code { + Some(code) => format!("pane %{pane_id} exited with code {code}"), + None => format!("pane %{pane_id} exited"), + }, + ControlEvent::AgentStatus { pane_id, json } => { + format!("pane %{pane_id} agent status: {json}") + } + ControlEvent::Preempted { workspace, by } => { + format!("workspace {workspace} taken over by {by}") + } + ControlEvent::Layout { workspace, delta } => { + format!("workspace {workspace} layout: {delta:?}") + } + ControlEvent::LayoutResync => "layout resync".to_string(), + other => format!("{other:?}"), + } +} + +fn agents(backend: &mut dyn Backend) -> Result { + match backend.control(ControlRequest::AgentStates)? { + ReplyOk::AgentStates(states) => report( + output::agents_table(&states), + json!({ "agents": serde_json::to_value(&states)? }), + ), + other => bail!("the server answered AgentStates with {other:?}"), + } +} + +fn status(backend: &mut dyn Backend) -> Result { + match backend.control(ControlRequest::Status)? { + ReplyOk::Status(status) => { + report(output::status_lines(&status), serde_json::to_value(&status)?) + } + other => bail!("the server answered Status with {other:?}"), + } +} + +fn machine_ls(backend: &mut dyn Backend) -> Result { + match backend.control(ControlRequest::Routes)? { + ReplyOk::Routes(routes) => report( + output::routes_table(&routes), + json!({ "machines": serde_json::to_value(&routes)? }), + ), + other => bail!("the server answered Routes with {other:?}"), + } +} + +fn doctor(ctx: &Context, backend: &mut dyn Backend) -> Result { let mark = |v: &Option| match v { Some(value) => format!("set ({value})"), None => "missing".to_string(), }; - let rows = vec![ + let mut rows = vec![ vec![address::ENV_SOCKET.to_string(), mark(&ctx.socket)], vec![address::ENV_WS.to_string(), mark(&ctx.ws)], vec![address::ENV_PANE.to_string(), mark(&ctx.pane)], ]; + let mut server = json!({ "reachable": false }); + match backend.hello() { + Ok(hello) => { + let dialect_ok = hello.control_version == CONTROL_VERSION + && hello.protocol_version == PROTOCOL_VERSION; + let dialect = if dialect_ok { + format!("ok (control v{CONTROL_VERSION}, protocol v{PROTOCOL_VERSION})") + } else { + format!( + "MISMATCH (server speaks control v{} protocol v{}, this build \ + v{CONTROL_VERSION}/v{PROTOCOL_VERSION})", + hello.control_version, hello.protocol_version + ) + }; + rows.push(vec!["server".to_string(), format!("ok (build {})", hello.build)]); + rows.push(vec!["dialect".to_string(), dialect]); + let status = match backend.control(ControlRequest::Status)? { + ReplyOk::Status(status) => status, + other => bail!("the server answered Status with {other:?}"), + }; + rows.push(vec![ + "status".to_string(), + format!( + "pid {}, up {}s, {} panes", + status.pid, status.uptime_secs, status.panes + ), + ]); + let routes = match backend.control(ControlRequest::Routes)? { + ReplyOk::Routes(routes) => routes, + other => bail!("the server answered Routes with {other:?}"), + }; + let connected = routes.iter().filter(|r| r.connected).count(); + rows.push(vec![ + "machine links".to_string(), + format!("{} known, {connected} connected", routes.len()), + ]); + server = json!({ + "reachable": true, + "dialect_ok": dialect_ok, + "build": hello.build, + "status": serde_json::to_value(&status)?, + "routes": serde_json::to_value(&routes)?, + }); + } + Err(e) => { + rows.push(vec!["server".to_string(), format!("unreachable — {e:#}")]); + } + } let mut human = output::table(&["CHECK", "RESULT"], &rows); if ctx.socket.is_none() && ctx.pane.is_none() { - human.push_str("\nnot inside a tty7 shell — address commands need an explicit %pane/@tab/workspace\n"); + human.push_str( + "\nnot inside a tty7 shell — address commands need an explicit %pane/@tab/workspace\n", + ); } - human.push_str( - "\nsocket reachability, dialect handshake, config parse, version skew, agent hooks \ - and remote links are checked once the transport client lands\n", - ); report( human, json!({ @@ -488,9 +576,7 @@ fn doctor(ctx: &Context) -> Result { "workspace": ctx.ws.is_some(), "pane": ctx.pane.is_some(), }, - "pending": [ - "socket", "dialect", "config", "versions", "agent-hooks", "remote-links", - ], + "server": server, }), ) } @@ -569,7 +655,7 @@ mod tests { } #[test] - fn new_creates_workspace_first_tab_and_spawns_the_shell() { + fn new_spawns_first_and_seeds_the_tab_with_the_daemons_pane_id() { let mut backend = mock(); let created = Workspace::default(); backend @@ -582,7 +668,6 @@ mod tests { assert_eq!( backend.control_calls, vec![ - ControlRequest::MachineGet, ControlRequest::WorkspaceCreate { name: None, workspace: None, @@ -598,11 +683,12 @@ mod tests { }, tab: None, }, - ] + ], + "the daemon-assigned pane id (6) lands in the tree op, so the spawn came first" ); assert_eq!( backend.spawned, - vec![(6, created.id, Some("C:\\newproj".to_string()))], + vec![(created.id, Some("C:\\newproj".to_string()))], "the tree op alone leaves a dead pane — the shell must be spawned" ); assert_eq!(human(out), created.id.to_string()); @@ -745,7 +831,7 @@ mod tests { tab: None, } ); - assert_eq!(backend.spawned, vec![(6, api.id, Some("C:\\elsewhere".to_string()))]); + assert_eq!(backend.spawned, vec![(api.id, Some("C:\\elsewhere".to_string()))]); } #[test] @@ -777,7 +863,7 @@ mod tests { ], "the new pane inherits the split pane's cwd" ); - assert_eq!(backend.spawned, vec![(6, api, Some("C:\\proj".to_string()))]); + assert_eq!(backend.spawned, vec![(api, Some("C:\\proj".to_string()))]); assert_eq!(human(out), "%6", "the new pane address is the printed result"); } @@ -908,15 +994,11 @@ mod tests { } #[test] - fn the_missing_protocol_verbs_say_which_request_they_wait_for() { + fn the_still_missing_verbs_say_so_without_touching_the_wire() { for (args, needle) in [ - (vec!["tty7", "agents"], "AgentStates"), - (vec!["tty7", "status"], "Status"), - (vec!["tty7", "server", "status"], "Status"), - (vec!["tty7", "machine", "ls"], "Routes"), (vec!["tty7", "ws", "stop", "api"], "not implemented"), - (vec!["tty7", "server", "start"], "not implemented"), (vec!["tty7", "machine", "connect", "devbox"], "not implemented"), + (vec!["tty7", "machine", "disconnect", "devbox"], "not implemented"), ] { let mut backend = mock(); let err = execute(cli(&args), &Context::default(), &mut backend) @@ -933,17 +1015,77 @@ mod tests { } #[test] - fn doctor_reports_the_injected_context() { - let out = human(run_cli(&["tty7", "doctor"], &Context::default(), &mut mock())); + fn agents_status_and_machine_ls_are_single_aggregate_requests() { + use tty7_core::daemon::control::{RouteInfo, ServerStatus}; + + let mut backend = mock(); + backend.replies.push_back(ReplyOk::AgentStates(Vec::new())); + let out = run_cli(&["tty7", "agents"], &Context::default(), &mut backend); + assert_eq!(backend.control_calls, vec![ControlRequest::AgentStates]); + assert!(human(out).contains("no agents"), "an empty panel says so"); + + let mut backend = mock(); + backend.replies.push_back(ReplyOk::Status(ServerStatus { + pid: 4242, + uptime_secs: 61, + panes: 3, + control_version: CONTROL_VERSION, + protocol_version: PROTOCOL_VERSION, + build: "26.7.5".into(), + socket: "127.0.0.1:5555".into(), + })); + let out = run_cli(&["tty7", "status"], &Context::default(), &mut backend); + assert_eq!(backend.control_calls, vec![ControlRequest::Status]); + let rendered = human(out); + assert!(rendered.contains("4242"), "{rendered}"); + assert!(rendered.contains("61s"), "{rendered}"); + + let mut backend = mock(); + backend.replies.push_back(ReplyOk::Routes(vec![RouteInfo { + key: "me@build-box:22".into(), + kind: "ssh".into(), + connected: true, + }])); + let out = run_cli(&["tty7", "machine", "ls"], &Context::default(), &mut backend); + assert_eq!(backend.control_calls, vec![ControlRequest::Routes]); + let rendered = human(out); + assert!(rendered.contains("local"), "machine 0 is always listed: {rendered}"); + assert!(rendered.contains("me@build-box:22"), "{rendered}"); + } + + fn doctor_backend() -> MockBackend { + use tty7_core::daemon::control::ServerStatus; + + let mut backend = mock(); + backend.replies.push_back(ReplyOk::Status(ServerStatus { + pid: 4242, + uptime_secs: 61, + panes: 3, + control_version: CONTROL_VERSION, + protocol_version: PROTOCOL_VERSION, + build: "26.7.5".into(), + socket: "127.0.0.1:5555".into(), + })); + backend.replies.push_back(ReplyOk::Routes(Vec::new())); + backend + } + + #[test] + fn doctor_reports_the_injected_context_and_the_server_half() { + let out = human(run_cli(&["tty7", "doctor"], &Context::default(), &mut doctor_backend())); assert!(out.contains("TTY7_SOCKET"), "{out}"); assert!(out.contains("missing"), "{out}"); + assert!(out.contains("dialect"), "{out}"); + assert!(out.contains(&format!("control v{CONTROL_VERSION}")), "{out}"); + assert!(out.contains("pid 4242"), "{out}"); + assert!(out.contains("0 known"), "{out}"); let ctx = Context { pane: Some("7".into()), ws: None, socket: Some("sock".into()), }; - let out = human(run_cli(&["tty7", "doctor"], &ctx, &mut mock())); + let out = human(run_cli(&["tty7", "doctor"], &ctx, &mut doctor_backend())); assert!(out.contains("set (sock)"), "{out}"); } } diff --git a/crates/tty7-cli/src/main.rs b/crates/tty7-cli/src/main.rs index 090c7db2..e8e75c1a 100644 --- a/crates/tty7-cli/src/main.rs +++ b/crates/tty7-cli/src/main.rs @@ -4,6 +4,7 @@ mod cli; mod commands; mod output; mod resolve; +mod server; #[cfg(test)] mod testbed; @@ -14,7 +15,7 @@ fn main() -> std::process::ExitCode { let json = cli.json; let quiet = cli.quiet; let ctx = address::Context::from_env(); - let mut backend = backend::StubBackend; + let mut backend = backend::RealBackend::new(cli.machine.clone()); match commands::execute(cli, &ctx, &mut backend) { Ok(commands::Outcome::Exit(code)) => std::process::exit(code), Ok(commands::Outcome::Report(report)) => { diff --git a/crates/tty7-cli/src/output.rs b/crates/tty7-cli/src/output.rs index f6411fcd..427809c2 100644 --- a/crates/tty7-cli/src/output.rs +++ b/crates/tty7-cli/src/output.rs @@ -1,5 +1,6 @@ use tty7_core::core::machine::{Machine, PaneNode, Workspace}; use tty7_core::core::session::WorkspaceId; +use tty7_core::daemon::control::{PaneAgentState, RouteInfo, ServerStatus}; use tty7_core::daemon::protocol::PaneProcs; use crate::resolve; @@ -163,6 +164,60 @@ pub fn procs_tables(procs: &PaneProcs) -> String { out } +pub fn agents_table(states: &[PaneAgentState]) -> String { + if states.is_empty() { + return "no agents running\n".to_string(); + } + let rows: Vec> = states + .iter() + .map(|s| { + vec![ + format!("%{}", s.pane_id), + s.agent + .map(|a| format!("{a:?}").to_lowercase()) + .unwrap_or_else(|| "-".to_string()), + format!("{:?}", s.state.status).to_lowercase(), + s.state.message.clone().unwrap_or_else(|| "-".to_string()), + ] + }) + .collect(); + table(&["PANE", "AGENT", "STATUS", "MESSAGE"], &rows) +} + +pub fn status_lines(status: &ServerStatus) -> String { + let rows = vec![ + vec!["pid".to_string(), status.pid.to_string()], + vec!["uptime".to_string(), format!("{}s", status.uptime_secs)], + vec!["panes".to_string(), status.panes.to_string()], + vec![ + "dialect".to_string(), + format!( + "control v{}, protocol v{}", + status.control_version, status.protocol_version + ), + ], + vec!["build".to_string(), status.build.clone()], + vec!["socket".to_string(), status.socket.clone()], + ]; + table(&["SERVER", ""], &rows) +} + +pub fn routes_table(routes: &[RouteInfo]) -> String { + let mut rows = vec![vec![ + "local".to_string(), + "local".to_string(), + "yes".to_string(), + ]]; + rows.extend(routes.iter().map(|r| { + vec![ + r.key.clone(), + r.kind.clone(), + if r.connected { "yes" } else { "no" }.to_string(), + ] + })); + table(&["MACHINE", "KIND", "CONNECTED"], &rows) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/tty7-cli/src/resolve.rs b/crates/tty7-cli/src/resolve.rs index 73d653de..a59c7de4 100644 --- a/crates/tty7-cli/src/resolve.rs +++ b/crates/tty7-cli/src/resolve.rs @@ -103,18 +103,6 @@ pub fn workspace_of_pane(machine: &Machine, pane: u64) -> Result<&Workspace> { .ok_or_else(|| anyhow::anyhow!("no pane %{pane} on this machine — `tty7 pane ls` lists them")) } -pub fn next_pane_id(machine: &Machine) -> u64 { - let recorded = machine.panes.iter().map(|p| p.id).max().unwrap_or(0); - let in_trees = machine - .workspaces - .iter() - .flat_map(|ws| ws.tabs.iter()) - .flat_map(|tab| tab.root.pane_ids()) - .max() - .unwrap_or(0); - recorded.max(in_trees) + 1 -} - pub fn short_id(id: &WorkspaceId) -> String { id.to_string().chars().take(8).collect() } @@ -184,10 +172,4 @@ mod tests { assert!(err.contains("%99"), "{err}"); } - #[test] - fn new_pane_ids_never_collide_with_records_or_trees() { - let m = two_workspace_machine(); - assert_eq!(next_pane_id(&m), 6, "highest existing pane is %5"); - assert_eq!(next_pane_id(&Machine::default()), 1, "a fresh machine starts at %1"); - } } diff --git a/crates/tty7-cli/src/server.rs b/crates/tty7-cli/src/server.rs new file mode 100644 index 00000000..7b5f8f68 --- /dev/null +++ b/crates/tty7-cli/src/server.rs @@ -0,0 +1,177 @@ +use std::path::PathBuf; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use anyhow::{Result, bail}; +use serde_json::json; +use tty7_core::client::PaneClient; +use tty7_core::core::config; +use tty7_core::daemon::spawn; + +use crate::commands::{Outcome, Report}; + +const START_TIMEOUT: Duration = Duration::from_secs(10); +const POLL_INTERVAL: Duration = Duration::from_millis(50); +const LOG_TAIL_LINES: usize = 40; + +pub const SERVER_EXE_ENV: &str = "TTY7_SERVER_EXE"; + +fn report(human: impl Into, json: serde_json::Value) -> Result { + Ok(Outcome::Report(Report { + human: human.into(), + json, + })) +} + +fn running() -> bool { + PaneClient::local().version().is_ok() +} + +pub fn start() -> Result { + if running() { + return report( + "the server is already running", + json!({ "started": false, "running": true }), + ); + } + let exe = server_exe()?; + let mut cmd = Command::new(&exe); + cmd.arg("--daemon"); + if let Some(dir) = config::config_dir_path() { + cmd.arg("--config-dir").arg(dir); + } + cmd.stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + detach(&mut cmd); + let child = cmd + .spawn() + .map_err(|e| anyhow::anyhow!("could not start {}: {e}", exe.display()))?; + let pid = child.id(); + let deadline = Instant::now() + START_TIMEOUT; + while !running() { + if Instant::now() >= deadline { + bail!( + "{} (pid {pid}) did not open its endpoints within {START_TIMEOUT:?}", + exe.display() + ); + } + std::thread::sleep(POLL_INTERVAL); + } + report( + format!("started {} (pid {pid})", exe.display()), + json!({ "started": true, "pid": pid, "exe": exe.display().to_string() }), + ) +} + +pub fn stop() -> Result { + if !running() { + return report( + "the server is not running", + json!({ "stopped": false, "running": false }), + ); + } + spawn::stop(); + if running() { + bail!("the server did not shut down on request"); + } + report("stopped", json!({ "stopped": true })) +} + +pub fn restart() -> Result { + if running() { + spawn::stop(); + if running() { + bail!("the server did not shut down on request"); + } + } + start() +} + +pub fn logs() -> Result { + let Some(path) = config::config_path("tty7.log") else { + bail!("no config directory, so no log file location"); + }; + let mut human = format!("{}\n", path.display()); + let mut lines: Vec = Vec::new(); + match std::fs::read_to_string(&path) { + Ok(contents) => { + lines = contents + .lines() + .rev() + .take(LOG_TAIL_LINES) + .map(str::to_string) + .collect(); + lines.reverse(); + for line in &lines { + human.push_str(line); + human.push('\n'); + } + } + Err(_) => { + human.push_str("no log file yet — set TTY7_LOG=info before starting the server\n"); + } + } + report( + human, + json!({ "path": path.display().to_string(), "lines": lines }), + ) +} + +fn server_exe() -> Result { + if let Some(explicit) = std::env::var_os(SERVER_EXE_ENV).filter(|v| !v.is_empty()) { + return Ok(PathBuf::from(explicit)); + } + let name = if cfg!(windows) { + "tty7-server.exe" + } else { + "tty7-server" + }; + if let Ok(own) = std::env::current_exe() { + if let Some(dir) = own.parent() { + let sibling = dir.join(name); + if sibling.exists() { + return Ok(sibling); + } + } + } + if let Some(paths) = std::env::var_os("PATH") { + for dir in std::env::split_paths(&paths) { + let candidate = dir.join(name); + if candidate.is_file() { + return Ok(candidate); + } + } + } + bail!( + "could not find {name} next to this binary or on PATH — install it, or point \ + {SERVER_EXE_ENV} at it" + ) +} + +#[cfg(unix)] +fn detach(cmd: &mut Command) { + use std::os::unix::process::CommandExt as _; + unsafe { + cmd.pre_exec(|| { + if libc::setsid() == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } +} + +#[cfg(windows)] +fn detach(cmd: &mut Command) { + use std::os::windows::process::CommandExt as _; + + const DETACHED_PROCESS: u32 = 0x0000_0008; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; + const CREATE_NO_WINDOW: u32 = 0x0800_0000; + + cmd.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW); +} + +#[cfg(not(any(unix, windows)))] +fn detach(_cmd: &mut Command) {} diff --git a/crates/tty7-cli/tests/cli_e2e.rs b/crates/tty7-cli/tests/cli_e2e.rs new file mode 100644 index 00000000..84d858e1 --- /dev/null +++ b/crates/tty7-cli/tests/cli_e2e.rs @@ -0,0 +1,290 @@ +use std::io::BufRead as _; +use std::path::PathBuf; +use std::process::{Child, Command, Output, Stdio}; +use std::time::{Duration, Instant}; + +use tty7_core::client::{ControlClient, PaneClient}; +use tty7_core::daemon::control::ControlHello; +use tty7_core::daemon::protocol::PROTOCOL_VERSION; + +const DAEMON_ENV: &str = "TTY7_CLI_E2E_DAEMON"; +const READY_WITHIN: Duration = Duration::from_secs(30); +const SETTLE_WITHIN: Duration = Duration::from_secs(60); + +fn main() { + if std::env::var(DAEMON_ENV).as_deref() == Ok("1") { + if let Err(e) = tty7_core::daemon::server::run_daemon() { + eprintln!("e2e daemon exited with error: {e}"); + std::process::exit(1); + } + return; + } + + let tests: &[(&str, fn(&Daemon))] = &[ + ("ls_on_an_empty_server", ls_on_an_empty_server), + ("new_builds_a_workspace_with_a_live_pane", new_builds_a_workspace_with_a_live_pane), + ("run_streams_output_and_passes_the_exit_code", run_streams_output_and_passes_the_exit_code), + ("send_then_capture_round_trip", send_then_capture_round_trip), + ("status_reports_the_live_server", status_reports_the_live_server), + ("events_stream_reports_a_workspace_creation", events_stream_reports_a_workspace_creation), + ]; + + let mut failed = 0; + for (name, test) in tests { + let daemon = Daemon::start(); + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| test(&daemon))); + drop(daemon); + match outcome { + Ok(()) => println!("test {name} ... ok"), + Err(_) => { + failed += 1; + println!("test {name} ... FAILED"); + } + } + } + if failed > 0 { + eprintln!("{failed} e2e test(s) failed"); + std::process::exit(1); + } +} + +struct Daemon { + child: Child, + dir: tempfile::TempDir, +} + +impl Daemon { + fn start() -> Daemon { + let dir = tempfile::TempDir::new().expect("a temp dir for the isolated server"); + let own = std::env::current_exe().expect("own test binary path"); + let child = Command::new(own) + .env(DAEMON_ENV, "1") + .env("TTY7_CONFIG_DIR", 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 the in-test tty7 server"); + let daemon = Daemon { child, dir }; + daemon.await_ready(); + daemon + } + + fn control_endpoint(&self) -> PathBuf { + let file = if cfg!(windows) { + "control.port" + } else { + "control.sock" + }; + self.dir.path().join(file) + } + + fn pane_endpoint(&self) -> PathBuf { + let file = if cfg!(windows) { + "daemon.port" + } else { + "daemon.sock" + }; + self.dir.path().join(file) + } + + fn await_ready(&self) { + let hello = ControlHello::host_rpc("e2e-probe", "e2e-probe"); + let deadline = Instant::now() + READY_WITHIN; + loop { + let control_up = ControlClient::connect_at(&self.control_endpoint(), &hello).is_ok(); + let panes_up = PaneClient::at(self.pane_endpoint()).version().is_ok(); + if control_up && panes_up { + return; + } + assert!( + Instant::now() < deadline, + "the isolated server did not open its endpoints within {READY_WITHIN:?}" + ); + std::thread::sleep(Duration::from_millis(50)); + } + } + + fn cli(&self, args: &[&str]) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_tty7")); + cmd.args(args) + .env("TTY7_CONFIG_DIR", self.dir.path()) + .env("TTY7_DATA_DIR", self.dir.path()) + .env("TTY7_CONTROL_SOCK", self.dir.path().join("control.sock")) + .env_remove("TTY7_PANE") + .env_remove("TTY7_WS") + .env_remove("TTY7_SOCKET"); + cmd + } + + fn run(&self, args: &[&str]) -> Output { + self.cli(args) + .output() + .unwrap_or_else(|e| panic!("could not run tty7 {args:?}: {e}")) + } + + fn run_ok(&self, args: &[&str]) -> String { + let out = self.run(args); + assert!( + out.status.success(), + "tty7 {args:?} failed ({}): {}{}", + out.status, + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr), + ); + String::from_utf8_lossy(&out.stdout).into_owned() + } + + fn run_json(&self, args: &[&str]) -> serde_json::Value { + let mut with_json: Vec<&str> = args.to_vec(); + with_json.push("--json"); + let out = self.run_ok(&with_json); + serde_json::from_str(&out) + .unwrap_or_else(|e| panic!("tty7 {args:?} --json printed no JSON ({e}): {out}")) + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn workdir() -> String { + std::env::temp_dir().display().to_string() +} + +fn one_shot(command: &str) -> Vec { + if cfg!(windows) { + vec!["cmd.exe".into(), "/d".into(), "/c".into(), command.into()] + } else { + vec!["/bin/sh".into(), "-c".into(), command.into()] + } +} + +fn ls_on_an_empty_server(daemon: &Daemon) { + let out = daemon.run_ok(&["ls"]); + assert!(out.contains("no workspaces"), "{out}"); +} + +fn new_builds_a_workspace_with_a_live_pane(daemon: &Daemon) { + let created = daemon.run_json(&["new", &workdir()]); + let ws_id = created["id"].as_str().expect("new prints the workspace id"); + let pane = created["pane"].as_u64().expect("new prints the pane id"); + assert!(pane >= 1, "the daemon names panes from 1, got {pane}"); + + let listed = daemon.run_json(&["ls"]); + let workspaces = listed["workspaces"].as_array().expect("ls --json lists workspaces"); + assert_eq!(workspaces.len(), 1, "{listed}"); + assert_eq!(workspaces[0]["id"].as_str(), Some(ws_id), "{listed}"); + assert_eq!(workspaces[0]["panes"].as_u64(), Some(1), "{listed}"); + + let panes = daemon.run_ok(&["pane", "ls"]); + assert!(panes.contains(&format!("%{pane}")), "{panes}"); +} + +fn run_streams_output_and_passes_the_exit_code(daemon: &Daemon) { + let echo = one_shot("echo tty7_e2e_run_marker"); + let mut args: Vec<&str> = vec!["run", "--"]; + args.extend(echo.iter().map(String::as_str)); + let out = daemon.run(&args); + assert!( + out.status.success(), + "run failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + let stdout = String::from_utf8_lossy(&out.stdout); + assert!(stdout.contains("tty7_e2e_run_marker"), "{stdout}"); + + let exit = one_shot("exit 7"); + let mut args: Vec<&str> = vec!["run", "--"]; + args.extend(exit.iter().map(String::as_str)); + let out = daemon.run(&args); + assert_eq!( + out.status.code(), + Some(7), + "the child's exit code must pass through: {}", + String::from_utf8_lossy(&out.stderr) + ); +} + +fn send_then_capture_round_trip(daemon: &Daemon) { + let created = daemon.run_json(&["new", &workdir()]); + let pane = created["pane"].as_u64().expect("new prints the pane id"); + let address = format!("%{pane}"); + + daemon.run_ok(&["send", &address, "echo tty7_e2e_capture_marker", "--enter"]); + + let deadline = Instant::now() + SETTLE_WITHIN; + loop { + let seen = daemon.run_ok(&["capture", &address, "--scrollback"]); + if seen.contains("tty7_e2e_capture_marker") { + return; + } + assert!( + Instant::now() < deadline, + "the sent text never showed up in the capture; last capture:\n{seen}" + ); + std::thread::sleep(Duration::from_millis(200)); + } +} + +fn status_reports_the_live_server(daemon: &Daemon) { + let status = daemon.run_json(&["status"]); + assert!(status["pid"].as_u64().is_some_and(|pid| pid > 0), "{status}"); + assert_eq!( + status["protocol_version"].as_u64(), + Some(u64::from(PROTOCOL_VERSION)), + "{status}" + ); + + let human = daemon.run_ok(&["server", "status"]); + assert!(human.contains("pid"), "{human}"); +} + +fn events_stream_reports_a_workspace_creation(daemon: &Daemon) { + let mut watcher = daemon + .cli(&["events", "--json"]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("start tty7 events"); + let stdout = watcher.stdout.take().expect("piped stdout"); + let (tx, rx) = std::sync::mpsc::channel::(); + std::thread::spawn(move || { + let reader = std::io::BufReader::new(stdout); + for line in reader.lines() { + let Ok(line) = line else { break }; + if tx.send(line).is_err() { + break; + } + } + }); + + let deadline = Instant::now() + SETTLE_WITHIN; + let mut seen = Vec::new(); + let verdict = 'outer: loop { + if Instant::now() >= deadline { + break false; + } + daemon.run_ok(&["ws", "new", "evtws"]); + let round = Instant::now() + Duration::from_secs(5); + while let Ok(line) = rx.recv_timeout(round.saturating_duration_since(Instant::now())) { + let is_event = serde_json::from_str::(&line).is_ok(); + seen.push(line); + if is_event { + break 'outer true; + } + } + }; + let _ = watcher.kill(); + let _ = watcher.wait(); + assert!( + verdict, + "no event line arrived within {SETTLE_WITHIN:?}; saw {seen:?}" + ); +}