From 753d1dceeee737ccefcb151361af19e539366295 Mon Sep 17 00:00:00 2001 From: thomas Date: Fri, 31 Jul 2026 09:44:22 +0800 Subject: [PATCH] feat(core): pane observers, TTY7_* context env, and control aggregate queries Server-side gaps for the tty7 CLI (docs/cli-design.md, the additive tier): - Pane multi-subscriber: the single controlling subscriber keeps its preemption semantics, and a pane now also carries N read-only observers. A new pane-protocol frame `Observe { pane_id, size }` (kind 54) joins a connection as an observer: it gets the Snapshot replay, then Output/Exited/Size, and its Input/Resize is answered with an Error frame instead of reaching the shell. Observers never displace the GUI, never defer the dead-pane reap, and are pruned when their connection goes away. - Context env injection: every locally spawned shell now carries TTY7_PANE (its pane id), TTY7_WS (the workspace named in Spawn), and TTY7_SOCKET (the control endpoint path), next to the existing TTY7 marker. ClientMsg::Spawn gains an optional `workspace` field, carried by the SPAWN_OWNED frame with serde defaults. - Aggregate queries: ControlRequest::{AgentStates, Routes, Status} with ReplyOk::{AgentStates, Routes, Status}. AgentStates snapshots each pane's live agent session state through a new Services::panes directory wired from the daemon's registry; Routes lists the SshManager's held connections with key/kind/liveness; Status reports pid, uptime from a process-start instant, pane count, both dialect versions, build, and the control socket path. - Dialect bump: CONTROL_VERSION 3 -> 4, PROTOCOL_VERSION 4 -> 5, following the convention set by bed22d8 and 1792bb8/a4972d3 where every additive variant bumped the strict-equality handshake versions (and with them the tty7-server-c{control}p{protocol} install name). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv --- crates/tty7-core/src/daemon/control.rs | 84 +++++- crates/tty7-core/src/daemon/pane.rs | 383 +++++++++++++++++++++--- crates/tty7-core/src/daemon/protocol.rs | 85 +++++- crates/tty7-core/src/daemon/server.rs | 164 +++++++++- crates/tty7-core/src/daemon/ssh/mod.rs | 58 ++++ crates/tty7-core/src/host/server.rs | 172 ++++++++++- crates/tty7-server/tests/routed_pane.rs | 2 + src/terminal/remote.rs | 1 + 8 files changed, 888 insertions(+), 61 deletions(-) diff --git a/crates/tty7-core/src/daemon/control.rs b/crates/tty7-core/src/daemon/control.rs index 39c76a70..62e66a9d 100644 --- a/crates/tty7-core/src/daemon/control.rs +++ b/crates/tty7-core/src/daemon/control.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use super::protocol::{MAX_FRAME, read_frame, write_frame}; -pub const CONTROL_VERSION: u32 = 3; +pub const CONTROL_VERSION: u32 = 4; const DIALECT_MARKER: &str = "speaks control v"; @@ -26,6 +26,11 @@ pub fn server_instance() -> &'static str { INSTANCE.get_or_init(|| uuid::Uuid::new_v4().to_string()) } +pub fn server_started() -> Instant { + static STARTED: OnceLock = OnceLock::new(); + *STARTED.get_or_init(Instant::now) +} + pub const WATCH_BURST_CAP: usize = 1024; pub const WATCH_COALESCE_WINDOW: Duration = Duration::from_millis(100); @@ -226,6 +231,38 @@ pub enum ControlRequest { old: u64, new: PaneSeed, }, + + AgentStates, + Routes, + Status, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PaneAgentState { + pub pane_id: u64, + #[serde(default)] + pub agent: Option, + pub state: crate::core::cli_agent::AgentSessionState, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RouteInfo { + pub key: String, + pub kind: String, + pub connected: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ServerStatus { + pub pid: u32, + pub uptime_secs: u64, + pub panes: u64, + pub control_version: u32, + pub protocol_version: u32, + #[serde(default)] + pub build: String, + #[serde(default)] + pub socket: String, } impl ControlRequest { @@ -265,6 +302,7 @@ impl ControlRequest { | PaneSetRatio { .. } | PaneMove { .. } | PaneReplace { .. } => Duration::from_secs(10), + AgentStates | Routes | Status => Duration::from_secs(5), } } @@ -314,6 +352,9 @@ pub enum ReplyOk { WorkspaceTree(Box), TabTree(Box), Panes(Vec), + AgentStates(Vec), + Routes(Vec), + Status(ServerStatus), } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -1358,6 +1399,9 @@ mod tests { dirs: vec!["/home/me/proj".into(), "/home/me/proj/src".into()], }, ControlRequest::WatchClose { id: 7 }, + ControlRequest::AgentStates, + ControlRequest::Routes, + ControlRequest::Status, ] } @@ -1404,6 +1448,34 @@ mod tests { stderr: vec![0x00, 0xff, 0xfe, b'\n'], })), ControlReply::Ok(ReplyOk::WatchId(42)), + ControlReply::Ok(ReplyOk::AgentStates(vec![PaneAgentState { + pane_id: 9, + agent: Some(crate::core::cli_agent::CLIAgent::Claude), + state: crate::core::cli_agent::AgentSessionState { + status: crate::core::cli_agent::AgentStatus::Waiting, + message: Some("needs permission".into()), + session_id: Some("sess-1".into()), + launch_argv: Some(vec!["claude".into()]), + rich: true, + cwd: Some("/work/api".into()), + activity: 3, + }, + }])), + ControlReply::Ok(ReplyOk::AgentStates(Vec::new())), + ControlReply::Ok(ReplyOk::Routes(vec![RouteInfo { + key: "me@build-box:22".into(), + kind: "ssh".into(), + connected: true, + }])), + ControlReply::Ok(ReplyOk::Status(ServerStatus { + pid: 4242, + uptime_secs: 61, + panes: 3, + control_version: CONTROL_VERSION, + protocol_version: crate::daemon::protocol::PROTOCOL_VERSION, + build: "26.7.5".into(), + socket: "/run/user/1000/tty7/daemon.sock".into(), + })), ControlReply::Err(WireError::new(WireErrorKind::NotFound, "no such file")), ControlReply::Err(WireError::new( WireErrorKind::PermissionDenied, @@ -1470,6 +1542,13 @@ mod tests { assert_eq!(first, server_instance()); } + #[test] + fn the_server_start_instant_is_fixed_per_process() { + let first = server_started(); + assert_eq!(first, server_started(), "uptime needs one fixed origin"); + assert!(first.elapsed() >= Duration::ZERO); + } + #[test] fn a_hello_without_an_instance_still_decodes() { let json = r#"{"control_version":2,"protocol_version":3,"build":"26.7.6", @@ -2030,6 +2109,9 @@ mod tests { }, s(20), ), + (R::AgentStates, s(5)), + (R::Routes, s(5)), + (R::Status, s(5)), ]; assert_eq!( cases.len(), diff --git a/crates/tty7-core/src/daemon/pane.rs b/crates/tty7-core/src/daemon/pane.rs index 33a07dfa..7b929ab5 100644 --- a/crates/tty7-core/src/daemon/pane.rs +++ b/crates/tty7-core/src/daemon/pane.rs @@ -112,13 +112,15 @@ struct SpawnConfig { } fn build_spawn_config( + pane: u64, cwd: Option, shell: Option, + workspace: Option<&str>, ) -> anyhow::Result { let initial_cwd = initial_working_directory(cwd); let configured = choose_shell(shell, crate::core::config::shell_command()); let remote = wsl_remote_context(configured.as_ref()); - let (cmd, integration_dir) = build_shell_command(configured, &initial_cwd)?; + let (cmd, integration_dir) = build_shell_command(configured, &initial_cwd, pane, workspace)?; Ok(SpawnConfig { cmd, initial_cwd, @@ -149,6 +151,8 @@ fn wsl_remote_context(shell: Option<&ChosenShell>) -> Option { fn build_shell_command( configured: Option, initial_cwd: &Option, + pane: u64, + workspace: Option<&str>, ) -> anyhow::Result<(CommandBuilder, Option)> { let mut cmd = match &configured { Some(chosen) => { @@ -172,7 +176,7 @@ fn build_shell_command( apply_shell_integration(&mut cmd, &resolved_program, integration); } let integration_dir = integration.as_ref().and_then(|i| i.dir.clone()); - apply_common_command_setup(&mut cmd, initial_cwd); + apply_common_command_setup(&mut cmd, initial_cwd, pane, workspace); Ok((cmd, integration_dir)) } @@ -260,6 +264,29 @@ fn system_locale_identifier() -> Option { const TERM_PROGRAM_NAME: &str = "tty7"; +const TTY7_SOCKET_ENV: &str = "TTY7_SOCKET"; +const TTY7_PANE_ENV: &str = "TTY7_PANE"; +const TTY7_WS_ENV: &str = "TTY7_WS"; + +#[cfg(unix)] +fn control_socket_env() -> Option { + crate::host::server::control_socket_path() + .ok() + .map(|p| p.display().to_string()) +} + +#[cfg(windows)] +fn control_socket_env() -> Option { + crate::host::server::control_endpoint_path() + .ok() + .map(|p| p.display().to_string()) +} + +#[cfg(not(any(unix, windows)))] +fn control_socket_env() -> Option { + None +} + const CAPABILITY_ENV: [&str; 2] = ["TERM", "COLORTERM"]; fn names_capability_env(key: &str) -> bool { @@ -274,6 +301,8 @@ fn names_capability_env(key: &str) -> bool { fn pane_environment( extra_env: &std::collections::HashMap, + pane: u64, + workspace: Option<&str>, ) -> Vec<(String, String)> { let version = env!("CARGO_PKG_VERSION"); let mut env = vec![ @@ -285,7 +314,14 @@ fn pane_environment( ), ("TERM_PROGRAM".to_string(), TERM_PROGRAM_NAME.to_string()), ("TERM_PROGRAM_VERSION".to_string(), version.to_string()), + (TTY7_PANE_ENV.to_string(), pane.to_string()), ]; + if let Some(ws) = workspace { + env.push((TTY7_WS_ENV.to_string(), ws.to_string())); + } + if let Some(socket) = control_socket_env() { + env.push((TTY7_SOCKET_ENV.to_string(), socket)); + } env.extend( extra_env .iter() @@ -295,12 +331,17 @@ fn pane_environment( env } -fn apply_common_command_setup(cmd: &mut CommandBuilder, initial_cwd: &Option) { +fn apply_common_command_setup( + cmd: &mut CommandBuilder, + initial_cwd: &Option, + pane: u64, + workspace: Option<&str>, +) { if let Some(dir) = initial_cwd { cmd.cwd(dir); } let extra_env = crate::core::config::extra_env(); - for (k, v) in pane_environment(&extra_env) { + for (k, v) in pane_environment(&extra_env, pane, workspace) { cmd.env(k, v); } @@ -379,6 +420,8 @@ struct PaneState { ring: ReplayRing, subscriber: Option>, subscriber_epoch: u64, + observers: Vec<(u64, Sender)>, + observer_seq: u64, cwd: Option, shell: ShellState, remote: Option, @@ -388,6 +431,13 @@ struct PaneState { alive: bool, } +fn notify(st: &mut PaneState, msg: DaemonMsg) { + if let Some(sub) = &st.subscriber { + let _ = sub.send(msg.clone()); + } + st.observers.retain(|(_, tx)| tx.send(msg.clone()).is_ok()); +} + enum PaneBackend { Pty(PtyBackend), NativeSsh(NativeSshBackend), @@ -450,9 +500,7 @@ impl DeathReporter { return; } let subscribed = st.subscriber.is_some(); - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Exited { code: None }); - } + notify(&mut st, DaemonMsg::Exited { code: None }); drop(st); crate::core::machine::observe_pane(pane, |p| p.live = false); if subscribed { @@ -471,12 +519,13 @@ impl DaemonPane { size: WinSize, shell: Option, owner: Option, + workspace: Option, on_dead: impl FnOnce() + Send + 'static, ) -> anyhow::Result> { let pty_size = pty_size(size); let pair = native_pty_system().openpty(pty_size)?; - let spawn = build_spawn_config(cwd, shell)?; + let spawn = build_spawn_config(id, cwd, shell, workspace.as_deref())?; let child = pair.slave.spawn_command(spawn.cmd)?; let shell_pid = child.process_id(); @@ -491,6 +540,8 @@ impl DaemonPane { ring: ReplayRing::new(size), subscriber: None, subscriber_epoch: 0, + observers: Vec::new(), + observer_seq: 0, cwd: spawn.initial_cwd, shell: ShellState::default(), remote: spawn.remote.clone(), @@ -579,6 +630,8 @@ impl DaemonPane { ring: ReplayRing::new(size), subscriber: None, subscriber_epoch: 0, + observers: Vec::new(), + observer_seq: 0, cwd: None, shell: ShellState::default(), remote: Some(remote), @@ -759,6 +812,9 @@ impl DaemonPane { gate.add(n); } } + st.observers.retain(|(_, tx)| { + tx.send(DaemonMsg::Output(bytes.to_vec())).is_ok() + }); apply_signals(&mut st, signals); if let Some(remote) = remote { apply_remote_context(&mut st, remote); @@ -816,6 +872,20 @@ impl DaemonPane { !st.alive && st.subscriber.is_none() } + pub fn observe(&self, observer: Sender) -> u64 { + let mut st = self.state.lock().unwrap(); + observe_subscriber(&mut st, observer) + } + + pub fn unobserve(&self, observer_id: u64) { + let mut st = self.state.lock().unwrap(); + st.observers.retain(|(id, _)| *id != observer_id); + } + + pub fn agent_state(&self) -> Option { + agent_state_snapshot(&self.state.lock().unwrap()) + } + pub fn gate(&self) -> Arc { self.gate.clone() } @@ -848,7 +918,12 @@ impl DaemonPane { } pub fn resize(&self, size: WinSize) { - self.state.lock().unwrap().ring.resize(size); + { + let mut st = self.state.lock().unwrap(); + st.ring.resize(size); + st.observers + .retain(|(_, tx)| tx.send(DaemonMsg::Size(size)).is_ok()); + } match &self.backend { PaneBackend::Pty(p) => { if let Ok(master) = p.master.lock() { @@ -1160,10 +1235,8 @@ impl ReplayRing { } } -fn attach_subscriber(st: &mut PaneState, subscriber: Sender) -> u64 { - st.subscriber_epoch += 1; - - st.ring.replay(&subscriber); +fn replay_state(st: &PaneState, subscriber: &Sender) { + st.ring.replay(subscriber); if let Some(cwd) = &st.cwd { let _ = subscriber.send(DaemonMsg::Cwd(cwd.clone())); } @@ -1186,10 +1259,32 @@ fn attach_subscriber(st: &mut PaneState, subscriber: Sender) -> u64 { if !st.alive { let _ = subscriber.send(DaemonMsg::Exited { code: None }); } +} + +fn attach_subscriber(st: &mut PaneState, subscriber: Sender) -> u64 { + st.subscriber_epoch += 1; + replay_state(st, &subscriber); st.subscriber = Some(subscriber); st.subscriber_epoch } +fn observe_subscriber(st: &mut PaneState, observer: Sender) -> u64 { + st.observer_seq += 1; + replay_state(st, &observer); + st.observers.push((st.observer_seq, observer)); + st.observer_seq +} + +fn agent_state_snapshot(st: &PaneState) -> Option { + st.agent_session + .clone() + .map(|state| crate::daemon::control::PaneAgentState { + pane_id: st.id, + agent: st.agent, + state, + }) +} + fn observed_facts(st: &PaneState) -> (Option, Option) { let cwd = st.cwd.as_ref().map(|p| p.to_string_lossy().into_owned()); let agent = st.agent.map(|agent| crate::core::machine::AgentFacts { @@ -1228,9 +1323,7 @@ fn agent_facts_changed( fn apply_signals(st: &mut PaneState, signals: SniffSignals) { if let Some(cwd) = signals.cwd { if st.cwd.as_ref() != Some(&cwd) { - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Cwd(cwd.clone())); - } + notify(st, DaemonMsg::Cwd(cwd.clone())); st.cwd = Some(cwd); } } @@ -1244,13 +1337,14 @@ fn apply_signals(st: &mut PaneState, signals: SniffSignals) { ); } st.shell = shell.clone(); - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Prompt { + notify( + st, + DaemonMsg::Prompt { active: shell.active, at_prompt: shell.at_prompt, last_exit: shell.last_exit_code, - }); - } + }, + ); } apply_agent_signals(st, signals.agent_events, signals.notification); } @@ -1286,9 +1380,7 @@ fn apply_agent_signals( for event in &events { if st.agent.is_none() && event.agent.is_some() { st.agent = event.agent; - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Agent(st.agent)); - } + notify(st, DaemonMsg::Agent(st.agent)); } st.agent_session .get_or_insert_with(AgentSessionState::default) @@ -1312,10 +1404,8 @@ fn apply_agent_signals( sess.launch_argv = Some(argv.clone()); } - if st.agent_session != before - && let Some(sub) = &st.subscriber - { - let _ = sub.send(DaemonMsg::AgentStatus(st.agent_session.clone())); + if st.agent_session != before { + notify(st, DaemonMsg::AgentStatus(st.agent_session.clone())); } } @@ -1329,9 +1419,7 @@ fn apply_probed_cwd(st: &mut PaneState, probed: Option) { if st.cwd.as_deref().is_some_and(|cur| same_dir(cur, &probed)) { return; } - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Cwd(probed.clone())); - } + notify(st, DaemonMsg::Cwd(probed.clone())); st.cwd = Some(probed); } @@ -1348,9 +1436,7 @@ fn apply_remote_context(st: &mut PaneState, remote: Option) { return; } st.cwd = None; - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::RemoteContext(remote.clone())); - } + notify(st, DaemonMsg::RemoteContext(remote.clone())); st.remote = remote; } @@ -1368,16 +1454,12 @@ fn apply_agent( } if agent.is_none() && st.agent_session.is_some() { st.agent_session = None; - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::AgentStatus(None)); - } + notify(st, DaemonMsg::AgentStatus(None)); } if agent.is_none() { st.agent_argv = None; } - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::Agent(agent)); - } + notify(st, DaemonMsg::Agent(agent)); st.agent = agent; stamp_launch_argv(st, argv); } @@ -1395,9 +1477,7 @@ fn stamp_launch_argv(st: &mut PaneState, argv: Option>) { && sess.launch_argv.as_ref() != Some(&argv) { sess.launch_argv = Some(argv); - if let Some(sub) = &st.subscriber { - let _ = sub.send(DaemonMsg::AgentStatus(st.agent_session.clone())); - } + notify(st, DaemonMsg::AgentStatus(st.agent_session.clone())); } } @@ -1756,6 +1836,7 @@ mod tests { args_are_tty7_defaults: false, }), None, + None, || {}, ) .expect("spawn pane"); @@ -2501,6 +2582,8 @@ mod tests { ring: ReplayRing::new(ws(80, 24)), subscriber: None, subscriber_epoch: 0, + observers: Vec::new(), + observer_seq: 0, cwd: None, shell: ShellState::default(), remote: None, @@ -2843,6 +2926,158 @@ mod tests { )); } + fn drain(rx: &mpsc::Receiver) -> Vec { + let mut got = Vec::new(); + while let Ok(msg) = rx.try_recv() { + got.push(msg); + } + got + } + + #[test] + fn observe_replays_state_without_displacing_the_controller() { + let mut st = test_state(true); + st.ring.append(b"screen"); + st.cwd = Some(PathBuf::from("/work")); + + let (controller_tx, controller_rx) = mpsc::channel(); + let epoch = attach_subscriber(&mut st, controller_tx); + drain(&controller_rx); + + let (observer_tx, observer_rx) = mpsc::channel(); + let id = observe_subscriber(&mut st, observer_tx); + 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!( + controller_rx.try_recv().is_err(), + "an observer joining must be invisible to the controller" + ); + + notify(&mut st, DaemonMsg::Output(b"tick".to_vec())); + assert!(matches!(controller_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick")); + assert!(matches!(observer_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"tick")); + + st.observers.retain(|(oid, _)| *oid != id); + 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"); + } + + #[test] + fn a_new_controller_preempts_the_old_while_observers_survive() { + let mut st = test_state(true); + + let (first_tx, first_rx) = mpsc::channel(); + let first_epoch = attach_subscriber(&mut st, first_tx); + drain(&first_rx); + + let (observer_tx, observer_rx) = mpsc::channel(); + observe_subscriber(&mut st, observer_tx); + drain(&observer_rx); + + let (second_tx, second_rx) = mpsc::channel(); + let second_epoch = attach_subscriber(&mut st, second_tx); + assert!(second_epoch > first_epoch); + drain(&second_rx); + + notify(&mut st, DaemonMsg::Output(b"live".to_vec())); + assert!( + matches!(first_rx.try_recv(), Err(mpsc::TryRecvError::Disconnected)), + "the preempted controller's channel must be gone, exactly as before" + ); + assert!(matches!(second_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"live")); + assert!( + matches!(observer_rx.try_recv(), Ok(DaemonMsg::Output(b)) if b == b"live"), + "a controller handover must not evict read-only observers" + ); + + if st.subscriber_epoch == first_epoch { + st.subscriber = None; + } + assert!( + st.subscriber.is_some(), + "a stale epoch's detach must not unseat the new controller" + ); + } + + #[test] + fn a_gone_observer_is_pruned_on_the_next_broadcast() { + let mut st = test_state(true); + let (observer_tx, observer_rx) = mpsc::channel(); + observe_subscriber(&mut st, observer_tx); + drop(observer_rx); + + notify(&mut st, DaemonMsg::Output(b"x".to_vec())); + assert!(st.observers.is_empty(), "a dead observer must not accumulate"); + } + + #[test] + fn death_notifies_observers_but_only_controllers_defer_the_reap() { + let with_observer_only = Arc::new(Mutex::new(test_state(true))); + let (observer_tx, observer_rx) = mpsc::channel(); + observe_subscriber(&mut with_observer_only.lock().unwrap(), observer_tx); + drain(&observer_rx); + 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!( + dead_rx.try_recv().is_ok(), + "read-only observers must not keep a dead pane in the registry" + ); + + let with_both = Arc::new(Mutex::new(test_state(true))); + let (controller_tx, controller_rx) = mpsc::channel(); + let (observer_tx, observer_rx) = mpsc::channel(); + { + let mut st = with_both.lock().unwrap(); + attach_subscriber(&mut st, controller_tx); + observe_subscriber(&mut st, observer_tx); + } + drain(&controller_rx); + drain(&observer_rx); + 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!( + dead_rx.try_recv().is_err(), + "an attached death is still the detach path's to reclaim" + ); + } + + #[test] + fn agent_state_snapshot_reports_only_panes_with_a_session() { + use crate::core::cli_agent::{AgentSessionState, AgentStatus, CLIAgent}; + + let mut st = test_state(true); + st.id = 42; + assert_eq!(agent_state_snapshot(&st), None); + + st.agent = Some(CLIAgent::Claude); + assert_eq!( + agent_state_snapshot(&st), + None, + "a detected agent without session state is not yet a fact worth listing" + ); + + st.agent_session = Some(AgentSessionState { + status: AgentStatus::Waiting, + session_id: Some("sess-1".into()), + ..Default::default() + }); + let snap = agent_state_snapshot(&st).expect("a session state is the fact"); + assert_eq!(snap.pane_id, 42); + assert_eq!(snap.agent, Some(CLIAgent::Claude)); + assert_eq!(snap.state.status, AgentStatus::Waiting); + assert_eq!(snap.state.session_id.as_deref(), Some("sess-1")); + } + #[test] fn reader_eof_with_subscriber_sends_exited_not_on_dead() { let state = Arc::new(Mutex::new(test_state(true))); @@ -3008,7 +3243,7 @@ mod tests { #[test] fn pane_environment_advertises_the_terminal_under_the_standard_names() { let env: std::collections::HashMap<_, _> = - pane_environment(&std::collections::HashMap::new()) + pane_environment(&std::collections::HashMap::new(), 7, Some("ws-main")) .into_iter() .collect(); let version = env!("CARGO_PKG_VERSION"); @@ -3030,6 +3265,45 @@ mod tests { ); } + #[test] + fn pane_environment_hands_the_shell_its_own_address() { + let env: std::collections::HashMap<_, _> = + pane_environment(&std::collections::HashMap::new(), 42, Some("ws-main")) + .into_iter() + .collect(); + + assert_eq!( + env.get(TTY7_PANE_ENV).map(String::as_str), + Some("42"), + "a CLI inside the pane needs its own pane id for address-free verbs" + ); + assert_eq!( + env.get(TTY7_WS_ENV).map(String::as_str), + Some("ws-main"), + "the workspace the spawn was filed under rides into the shell" + ); + match control_socket_env() { + Some(socket) => assert_eq!( + env.get(TTY7_SOCKET_ENV), + Some(&socket), + "the shell is told where this server answers" + ), + None => assert!( + !env.contains_key(TTY7_SOCKET_ENV), + "no resolvable endpoint must not inject an empty TTY7_SOCKET" + ), + } + + let unfiled: std::collections::HashMap<_, _> = + pane_environment(&std::collections::HashMap::new(), 42, None) + .into_iter() + .collect(); + assert!( + !unfiled.contains_key(TTY7_WS_ENV), + "a pane outside any workspace must not claim one" + ); + } + #[test] fn pane_environment_lets_configured_env_override_identity_but_not_capability() { let configured = [ @@ -3044,7 +3318,7 @@ mod tests { .collect(); let applied: std::collections::HashMap<_, _> = - pane_environment(&configured).into_iter().collect(); + pane_environment(&configured, 1, None).into_iter().collect(); assert_eq!( applied.get("TERM_PROGRAM").map(String::as_str), @@ -3073,7 +3347,7 @@ mod tests { .map(|(k, v)| ((*k).to_string(), (*v).to_string())) .collect(); - let applied = pane_environment(&configured); + let applied = pane_environment(&configured, 1, None); assert!( !applied.iter().any(|(k, _)| k == "Term" || k == "ColorTerm"), @@ -3179,7 +3453,7 @@ mod tests { #[test] fn spawned_shell_carries_the_tty7_marker() { - let cmd = build_shell_command(None, &Some(PathBuf::from("/tmp"))) + let cmd = build_shell_command(None, &Some(PathBuf::from("/tmp")), 42, Some("ws-main")) .expect("build default shell command") .0; let tty7 = cmd @@ -3190,6 +3464,23 @@ mod tests { Some(env!("CARGO_PKG_VERSION")), "the daemon must inject TTY7 into every spawned shell" ); + assert_eq!( + cmd.get_env(TTY7_PANE_ENV).and_then(|v| v.to_str()), + Some("42"), + "the daemon must tell every spawned shell which pane it is" + ); + assert_eq!( + cmd.get_env(TTY7_WS_ENV).and_then(|v| v.to_str()), + Some("ws-main"), + "the daemon must tell every spawned shell which workspace filed it" + ); + if let Some(socket) = control_socket_env() { + assert_eq!( + cmd.get_env(TTY7_SOCKET_ENV).and_then(|v| v.to_str()), + Some(socket.as_str()), + "the daemon must tell every spawned shell where its server answers" + ); + } } #[test] diff --git a/crates/tty7-core/src/daemon/protocol.rs b/crates/tty7-core/src/daemon/protocol.rs index 1d55c557..c6cf4fd2 100644 --- a/crates/tty7-core/src/daemon/protocol.rs +++ b/crates/tty7-core/src/daemon/protocol.rs @@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize}; pub const MAX_FRAME: usize = 64 * 1024 * 1024; -pub const PROTOCOL_VERSION: u32 = 4; +pub const PROTOCOL_VERSION: u32 = 5; pub const FEATURE_PANE_OWNER: &str = "pane-owner"; @@ -573,11 +573,16 @@ pub enum ClientMsg { size: WinSize, shell: Option, owner: Option, + workspace: Option, }, Attach { pane_id: u64, size: WinSize, }, + Observe { + pane_id: u64, + size: WinSize, + }, Input(Vec), Resize(WinSize), Detach, @@ -705,6 +710,7 @@ mod kind { pub const QUERY_PROCS: u8 = 50; pub const ON_WORKSPACE: u8 = 52; pub const SPAWN_OWNED: u8 = 53; + pub const OBSERVE: u8 = 54; pub const SPAWNED: u8 = 1; pub const SNAPSHOT: u8 = 2; @@ -809,6 +815,8 @@ struct OwnedSpawn { shell: Option, #[serde(default)] owner: Option, + #[serde(default)] + workspace: Option, } impl ClientMsg { @@ -819,18 +827,21 @@ impl ClientMsg { size, shell: None, owner: None, + workspace: None, } => write_frame(w, kind::SPAWN, &to_json(&(cwd, size))?), ClientMsg::Spawn { cwd, size, shell: shell @ Some(_), owner: None, + workspace: None, } => write_frame(w, kind::SPAWN_SHELL, &to_json(&(cwd, size, shell))?), ClientMsg::Spawn { cwd, size, shell, - owner: owner @ Some(_), + owner, + workspace, } => write_frame( w, kind::SPAWN_OWNED, @@ -839,11 +850,15 @@ impl ClientMsg { size: *size, shell: shell.clone(), owner: owner.clone(), + workspace: workspace.clone(), })?, ), ClientMsg::Attach { pane_id, size } => { write_frame(w, kind::ATTACH, &to_json(&(pane_id, size))?) } + ClientMsg::Observe { pane_id, size } => { + write_frame(w, kind::OBSERVE, &to_json(&(pane_id, size))?) + } ClientMsg::Input(bytes) => write_frame(w, kind::INPUT, bytes), ClientMsg::Resize(size) => write_frame(w, kind::RESIZE, &to_json(size)?), ClientMsg::Detach => write_frame(w, kind::DETACH, &[]), @@ -910,6 +925,7 @@ impl ClientMsg { size, shell: None, owner: None, + workspace: None, } } kind::SPAWN_SHELL => { @@ -919,6 +935,7 @@ impl ClientMsg { size, shell, owner: None, + workspace: None, } } kind::SPAWN_OWNED => { @@ -927,18 +944,24 @@ impl ClientMsg { size, shell, owner, + workspace, } = from_json(&payload)?; ClientMsg::Spawn { cwd, size, shell, owner, + workspace, } } kind::ATTACH => { let (pane_id, size) = from_json(&payload)?; ClientMsg::Attach { pane_id, size } } + kind::OBSERVE => { + let (pane_id, size) = from_json(&payload)?; + ClientMsg::Observe { pane_id, size } + } kind::INPUT => ClientMsg::Input(payload), kind::RESIZE => ClientMsg::Resize(from_json(&payload)?), kind::DETACH => ClientMsg::Detach, @@ -1149,6 +1172,7 @@ mod tests { size: SIZE, shell: None, owner: None, + workspace: None, }, ClientMsg::Resize(SIZE), ClientMsg::Input(vec![b'l', b's', b'\r']), @@ -1202,12 +1226,14 @@ mod tests { size: SIZE, shell: None, owner: None, + workspace: None, }, ClientMsg::Spawn { cwd: None, size: SIZE, shell: None, owner: None, + workspace: None, }, ClientMsg::Spawn { cwd: Some(PathBuf::from("/tmp/x")), @@ -1218,12 +1244,25 @@ mod tests { args_are_tty7_defaults: true, }), owner: None, + workspace: None, }, ClientMsg::Spawn { cwd: Some(PathBuf::from("/tmp/x")), size: SIZE, shell: None, owner: Some("bda10e44-02de-44a0-8412-ec1cda2b5f5b".into()), + workspace: None, + }, + ClientMsg::Spawn { + cwd: Some(PathBuf::from("/tmp/x")), + size: SIZE, + shell: None, + owner: None, + workspace: Some("ws-main".into()), + }, + ClientMsg::Observe { + pane_id: 42, + size: SIZE, }, ClientMsg::Attach { pane_id: 42, @@ -1513,6 +1552,7 @@ mod tests { size: SIZE, shell: None, owner: None, + workspace: None, }; let mut buf = Vec::new(); msg.encode(&mut buf).unwrap(); @@ -1531,6 +1571,7 @@ mod tests { size: SIZE, shell: None, owner: None, + workspace: None, } ); } @@ -1547,6 +1588,7 @@ mod tests { size: SIZE, shell: Some(shell.clone()), owner: None, + workspace: None, }; let mut buf = Vec::new(); msg.encode(&mut buf).unwrap(); @@ -1560,6 +1602,7 @@ mod tests { size: SIZE, shell: Some(shell), owner: None, + workspace: None, } ); } @@ -1575,6 +1618,7 @@ mod tests { args_are_tty7_defaults: false, }), owner: Some("bda10e44-02de-44a0-8412-ec1cda2b5f5b".into()), + workspace: Some("ws-7".into()), }; let mut buf = Vec::new(); msg.encode(&mut buf).unwrap(); @@ -1603,10 +1647,45 @@ mod tests { }, shell: None, owner: None, + workspace: None, } ); } + #[test] + fn a_workspace_spawn_uses_the_owned_kind_and_round_trips() { + let msg = ClientMsg::Spawn { + cwd: None, + size: SIZE, + shell: None, + owner: None, + workspace: Some("ws-main".into()), + }; + let mut buf = Vec::new(); + msg.encode(&mut buf).unwrap(); + let (k, payload) = read_frame(&mut std::io::Cursor::new(&buf)).unwrap(); + assert_eq!( + k, + kind::SPAWN_OWNED, + "a workspace-tagged spawn must not ride the legacy kinds, which drop the field" + ); + assert_eq!(ClientMsg::from_frame(k, payload).unwrap(), msg); + } + + #[test] + fn observe_uses_its_own_kind_and_round_trips() { + let msg = ClientMsg::Observe { + pane_id: 42, + size: SIZE, + }; + let mut buf = Vec::new(); + msg.encode(&mut buf).unwrap(); + let (k, payload) = read_frame(&mut std::io::Cursor::new(&buf)).unwrap(); + assert_eq!(k, kind::OBSERVE); + assert_ne!(k, kind::ATTACH, "observing must never preempt an attach"); + assert_eq!(ClientMsg::from_frame(k, payload).unwrap(), msg); + } + #[test] fn pane_info_owner_defaults_for_old_daemons() { let old = serde_json::json!({"pane_id": 3, "title": "zsh", "alive": true}); @@ -1993,7 +2072,7 @@ mod tests { #[test] fn the_local_daemon_does_not_claim_the_control_dialect() { let v = DaemonVersion::current(); - assert_eq!(v.protocol, 4); + assert_eq!(v.protocol, 5); assert!( !v.has_feature(crate::daemon::control::feature::CONTROL), "the session daemon must not advertise a dialect it cannot serve" diff --git a/crates/tty7-core/src/daemon/server.rs b/crates/tty7-core/src/daemon/server.rs index b8a39583..b39ae261 100644 --- a/crates/tty7-core/src/daemon/server.rs +++ b/crates/tty7-core/src/daemon/server.rs @@ -79,6 +79,19 @@ impl Registry { } } +impl crate::host::server::PaneDirectory for Registry { + fn pane_count(&self) -> u64 { + self.panes.lock().unwrap().len() as u64 + } + + fn agent_states(&self) -> Vec { + let panes: Vec> = self.panes.lock().unwrap().values().cloned().collect(); + let mut states: Vec<_> = panes.iter().filter_map(|p| p.agent_state()).collect(); + states.sort_by_key(|s| s.pane_id); + states + } +} + const ORPHAN_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(600); fn spawn_orphan_sweep(registry: Arc) { @@ -131,18 +144,24 @@ fn ssh_connection_for( } pub fn run_daemon() -> anyhow::Result<()> { + let registry = Arc::new(Registry::new()); + #[cfg(any(unix, windows))] - match crate::host::server::spawn_control_listener_with( - crate::host::local::LocalHost::shared(), - control_services(), - ) { - Ok(path) => eprintln!("tty7-server: control socket at {}", path.display()), - Err(e) => eprintln!("tty7-server: control listener unavailable: {e}"), + { + let mut services = control_services(); + services.panes = Some(registry.clone()); + match crate::host::server::spawn_control_listener_with( + crate::host::local::LocalHost::shared(), + services, + ) { + Ok(path) => eprintln!("tty7-server: control socket at {}", path.display()), + Err(e) => eprintln!("tty7-server: control listener unavailable: {e}"), + } } #[cfg(not(any(unix, windows)))] log::info!("no control listener on this platform; serving panes only"); - run() + run_with(registry) } pub fn control_services() -> crate::host::server::Services { @@ -161,6 +180,12 @@ pub fn control_services() -> crate::host::server::Services { } pub fn run() -> anyhow::Result<()> { + run_with(Arc::new(Registry::new())) +} + +fn run_with(registry: Arc) -> anyhow::Result<()> { + crate::daemon::control::server_started(); + if transport::endpoint_exists() { match transport::connect() { Ok(_) => { @@ -180,8 +205,6 @@ pub fn run() -> anyhow::Result<()> { crate::daemon::pidfile::write_current(); - let registry = Arc::new(Registry::new()); - #[cfg(unix)] serve_sigterm(registry.clone()); @@ -277,6 +300,7 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { size, shell, owner, + workspace, } => { let id = registry.alloc_id(); let on_dead = { @@ -290,7 +314,7 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { .ok(); } }; - let pane = match DaemonPane::spawn(id, cwd, size, shell, owner, on_dead) { + let pane = match DaemonPane::spawn(id, cwd, size, shell, owner, workspace, on_dead) { Ok(p) => p, Err(e) => { let mut w = write_stream; @@ -348,6 +372,15 @@ fn handle_conn(stream: Stream, registry: Arc) -> anyhow::Result<()> { } }, + ClientMsg::Observe { pane_id, size: _ } => match registry.get(pane_id) { + Some(pane) => stream_observer(pane, read_stream, write_stream), + None => { + let mut w = write_stream; + DaemonMsg::Error(format!("no such pane {pane_id}")).encode(&mut w)?; + Ok(()) + } + }, + ClientMsg::List => { let mut w = write_stream; DaemonMsg::PaneList(registry.list()).encode(&mut w)?; @@ -578,6 +611,46 @@ fn stream_pane( run_stream(pane, id, epoch, rx, read_stream, write_stream, registry) } +fn stream_observer( + pane: Arc, + mut read_stream: Stream, + write_stream: Stream, +) -> anyhow::Result<()> { + let (tx, rx) = mpsc::channel::(); + let refusals = tx.clone(); + let observer_id = pane.observe(tx); + let writer = spawn_writer( + rx, + write_stream, + Arc::new(crate::daemon::pane::OutputGate::new()), + ); + + observe_loop(&mut read_stream, &refusals); + + pane.unobserve(observer_id); + drop(refusals); + let _ = writer.join(); + Ok(()) +} + +fn observe_loop(read_stream: &mut R, refusals: &mpsc::Sender) { + loop { + match ClientMsg::read(read_stream) { + Ok(ClientMsg::Input(_)) | Ok(ClientMsg::Resize(_)) => { + let refused = refusals.send(DaemonMsg::Error( + "this connection is a read-only observer; attach to write".to_string(), + )); + if refused.is_err() { + break; + } + } + Ok(ClientMsg::Detach) => break, + Ok(_) => {} + Err(_) => break, + } + } +} + fn run_stream( pane: Arc, id: u64, @@ -726,6 +799,61 @@ mod tests { assert!(reg.list().is_empty()); } + #[test] + fn an_empty_registry_serves_empty_aggregates() { + use crate::host::server::PaneDirectory as _; + let reg = Registry::new(); + assert_eq!(reg.pane_count(), 0); + assert!(reg.agent_states().is_empty()); + } + + #[test] + fn the_observer_loop_refuses_writes_and_honors_detach() { + let size = crate::daemon::protocol::WinSize { + cols: 80, + rows: 24, + cell_w: 8, + cell_h: 17, + }; + let mut wire = Vec::new(); + ClientMsg::Input(b"echo hijack\r".to_vec()) + .encode(&mut wire) + .unwrap(); + ClientMsg::Resize(size).encode(&mut wire).unwrap(); + ClientMsg::QueryProcs { pane_id: 1 } + .encode(&mut wire) + .unwrap(); + ClientMsg::Detach.encode(&mut wire).unwrap(); + ClientMsg::Input(b"too late".to_vec()) + .encode(&mut wire) + .unwrap(); + + let (tx, rx) = std::sync::mpsc::channel(); + observe_loop(&mut std::io::Cursor::new(wire), &tx); + drop(tx); + + assert!( + matches!(rx.try_recv(), Ok(DaemonMsg::Error(m)) if m.contains("read-only")), + "an observer's Input must be answered with an Error, not forwarded" + ); + assert!( + matches!(rx.try_recv(), Ok(DaemonMsg::Error(m)) if m.contains("read-only")), + "an observer's Resize must be answered with an Error, not applied" + ); + assert!( + rx.try_recv().is_err(), + "Detach ends the loop; frames after it are never read" + ); + } + + #[test] + fn the_observer_loop_ends_at_stream_eof() { + let (tx, rx) = std::sync::mpsc::channel(); + observe_loop(&mut std::io::Cursor::new(Vec::::new()), &tx); + drop(tx); + assert!(rx.try_recv().is_err()); + } + #[cfg(unix)] mod conn { use super::super::{OUTPUT_COALESCE_CAP, Registry, handle_conn, spawn_writer}; @@ -777,6 +905,22 @@ mod tests { h.join().unwrap(); } + #[test] + fn observe_of_a_missing_pane_reports_error() { + let (mut client, h) = serve(); + ClientMsg::Observe { + pane_id: 999, + size: SIZE, + } + .encode(&mut client) + .unwrap(); + match DaemonMsg::read(&mut client).unwrap() { + DaemonMsg::Error(msg) => assert!(msg.contains("999"), "error names the id"), + other => panic!("expected Error, got {other:?}"), + } + h.join().unwrap(); + } + #[test] fn kill_unknown_pane_closes_without_reply() { let (mut client, h) = serve(); diff --git a/crates/tty7-core/src/daemon/ssh/mod.rs b/crates/tty7-core/src/daemon/ssh/mod.rs index 88e73ca5..9f465741 100644 --- a/crates/tty7-core/src/daemon/ssh/mod.rs +++ b/crates/tty7-core/src/daemon/ssh/mod.rs @@ -374,6 +374,26 @@ impl SshManager { self.conns.lock().unwrap().remove(key); } + pub fn routes(&self) -> Vec { + let conns = self.conns.lock().unwrap(); + 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); + crate::daemon::control::RouteInfo { + key: key.as_str().to_string(), + kind: "ssh".to_string(), + connected, + } + }) + .collect(); + routes.sort_by(|a, b| a.key.cmp(&b.key)); + routes + } + async fn remote_bootstrap(&self, conn: &Arc) -> Option { let key = conn.key().clone(); let cached = { self.probes.lock().unwrap().get(&key).cloned() }; @@ -633,6 +653,44 @@ mod tests { ); } + #[test] + fn routes_names_each_held_connection_with_its_liveness() { + 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()), + }; + assert!(mgr.routes().is_empty()); + + let mut other = base_spec(); + other.host = "build-box".into(); + for spec in [&base_spec(), &other] { + mgr.conns.lock().unwrap().insert( + ConnectionKey::from_spec(spec), + Arc::new(tokio::sync::Mutex::new(Weak::new())), + ); + } + + let routes = mgr.routes(); + let keys: Vec<&str> = routes.iter().map(|r| r.key.as_str()).collect(); + assert_eq!( + keys, + vec!["u@build-box:22", "u@h:22"], + "every held connection is listed, in a stable order" + ); + for route in &routes { + assert_eq!(route.kind, "ssh"); + assert!( + !route.connected, + "a dropped connection must read as disconnected, not vanish" + ); + } + } + #[test] fn connection_key_includes_jump_chain() { let mut with_jump = base_spec(); diff --git a/crates/tty7-core/src/host/server.rs b/crates/tty7-core/src/host/server.rs index 02cd32f1..61c6a5cf 100644 --- a/crates/tty7-core/src/host/server.rs +++ b/crates/tty7-core/src/host/server.rs @@ -9,7 +9,8 @@ use crate::core::machine::{self, Attachment, MachineStore}; use crate::daemon::control::{ CONTROL_VERSION, ControlClientMsg, ControlEvent, ControlHello, ControlHelloOk, ControlReply, ControlRequest, ControlServerMsg, GIT_STREAM_CHUNK, GIT_STREAM_CHUNK_MAX, LinkShutdown, - MAX_CONCURRENT_GIT_STREAMS, ReplyOk, WATCH_BURST_CAP, WireError, WireErrorKind, feature, + MAX_CONCURRENT_GIT_STREAMS, PaneAgentState, ReplyOk, ServerStatus, WATCH_BURST_CAP, WireError, + WireErrorKind, feature, server_started, }; use crate::daemon::duplex::{Duplex, Halves}; use crate::host::{Host, SearchHit, SharedHost, WatchSub}; @@ -22,10 +23,16 @@ pub const MAX_QUEUED: usize = 1024; pub const LAYOUT_EVENT_QUEUE: usize = 1024; +pub trait PaneDirectory: Send + Sync { + fn pane_count(&self) -> u64; + fn agent_states(&self) -> Vec; +} + #[derive(Clone, Default)] pub struct Services { pub machine: Option>, pub attachments: Arc, + pub panes: Option>, } impl Services { @@ -37,6 +44,7 @@ impl Services { Services { machine: Some(store), attachments: Arc::new(AttachRegistry::default()), + panes: None, } } } @@ -232,6 +240,7 @@ where machine: services.machine.clone(), machine_origin: machine_sub.as_ref().map(machine::Subscription::id), attachments: Arc::clone(&services.attachments), + panes: services.panes.clone(), id: NEXT_CONN.fetch_add(1, Ordering::Relaxed), holder: Holder { token: hello.client_token.clone(), @@ -722,9 +731,54 @@ fn run_request( .pane_replace(workspace, old, new, conn.machine_origin)?; (ReplyOk::Unit, Vec::new()) } + + ControlRequest::AgentStates => ( + ReplyOk::AgentStates( + conn.panes + .as_ref() + .map(|p| p.agent_states()) + .unwrap_or_default(), + ), + Vec::new(), + ), + ControlRequest::Routes => ( + ReplyOk::Routes(crate::daemon::ssh::SshManager::global().routes()), + Vec::new(), + ), + ControlRequest::Status => ( + ReplyOk::Status(ServerStatus { + pid: std::process::id(), + uptime_secs: server_started().elapsed().as_secs(), + panes: conn.panes.as_ref().map(|p| p.pane_count()).unwrap_or(0), + control_version: CONTROL_VERSION, + protocol_version: crate::daemon::protocol::PROTOCOL_VERSION, + build: env!("CARGO_PKG_VERSION").to_string(), + socket: control_endpoint_display(), + }), + Vec::new(), + ), }) } +#[cfg(unix)] +fn control_endpoint_display() -> String { + control_socket_path() + .map(|p| p.display().to_string()) + .unwrap_or_default() +} + +#[cfg(windows)] +fn control_endpoint_display() -> String { + control_endpoint_path() + .map(|p| p.display().to_string()) + .unwrap_or_default() +} + +#[cfg(not(any(unix, windows)))] +fn control_endpoint_display() -> String { + String::new() +} + fn paths(v: &[String]) -> Vec { v.iter().map(PathBuf::from).collect() } @@ -745,6 +799,7 @@ struct Conn { machine: Option>, machine_origin: Option, attachments: Arc, + panes: Option>, id: u64, holder: Holder, } @@ -1451,6 +1506,121 @@ pub use wsock::{ spawn_control_listener, spawn_control_listener_with, }; +#[cfg(test)] +mod aggregate_tests { + use super::*; + use crate::core::cli_agent::{AgentSessionState, AgentStatus, CLIAgent}; + use crate::daemon::control::ControlClient; + use crate::host::local::LocalHost; + use std::net::{TcpListener, TcpStream}; + + struct ThreePanesOneAgent; + + impl PaneDirectory for ThreePanesOneAgent { + fn pane_count(&self) -> u64 { + 3 + } + + fn agent_states(&self) -> Vec { + vec![PaneAgentState { + pane_id: 7, + agent: Some(CLIAgent::Claude), + state: AgentSessionState { + status: AgentStatus::Working, + session_id: Some("sess-7".into()), + ..Default::default() + }, + }] + } + } + + fn client_with(services: Services) -> ControlClient { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let _ = serve_with(stream, LocalHost::new(), services); + }); + let sock = TcpStream::connect(addr).unwrap(); + ControlClient::over_tcp( + sock, + &ControlHello::host_rpc("tok", "test-host"), + Box::new(|_| {}), + ) + .unwrap() + } + + #[test] + fn status_answers_with_this_servers_facts() { + let services = Services { + panes: Some(Arc::new(ThreePanesOneAgent)), + ..Services::none() + }; + let client = client_with(services); + + let ReplyOk::Status(status) = client.call(ControlRequest::Status).unwrap() else { + panic!("Status must answer with ReplyOk::Status"); + }; + assert_eq!(status.pid, std::process::id()); + assert_eq!(status.panes, 3); + assert_eq!(status.control_version, CONTROL_VERSION); + assert_eq!( + status.protocol_version, + crate::daemon::protocol::PROTOCOL_VERSION + ); + assert_eq!(status.build, env!("CARGO_PKG_VERSION")); + assert_eq!( + status.socket, + control_endpoint_display(), + "the CLI dials whatever path Status names" + ); + assert!(status.uptime_secs <= server_started().elapsed().as_secs()); + } + + #[test] + fn agent_states_are_the_pane_directorys_snapshot() { + let services = Services { + panes: Some(Arc::new(ThreePanesOneAgent)), + ..Services::none() + }; + let client = client_with(services); + + let ReplyOk::AgentStates(states) = client.call(ControlRequest::AgentStates).unwrap() + else { + panic!("AgentStates must answer with ReplyOk::AgentStates"); + }; + assert_eq!(states.len(), 1); + assert_eq!(states[0].pane_id, 7); + assert_eq!(states[0].agent, Some(CLIAgent::Claude)); + assert_eq!(states[0].state.status, AgentStatus::Working); + assert_eq!(states[0].state.session_id.as_deref(), Some("sess-7")); + } + + #[test] + 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 { + panic!("AgentStates must answer with ReplyOk::AgentStates"); + }; + assert!(states.is_empty()); + + let ReplyOk::Status(status) = client.call(ControlRequest::Status).unwrap() else { + panic!("Status must answer with ReplyOk::Status"); + }; + assert_eq!(status.panes, 0); + + let ReplyOk::Routes(routes) = client.call(ControlRequest::Routes).unwrap() else { + panic!("Routes must answer with ReplyOk::Routes"); + }; + assert!( + routes.iter().all(|r| !r.key.is_empty()), + "whatever links exist are named; none are blank" + ); + } +} + #[cfg(test)] mod pool_tests { use super::*; diff --git a/crates/tty7-server/tests/routed_pane.rs b/crates/tty7-server/tests/routed_pane.rs index 0953ce8b..1682d2bf 100644 --- a/crates/tty7-server/tests/routed_pane.rs +++ b/crates/tty7-server/tests/routed_pane.rs @@ -109,6 +109,7 @@ fn a_routed_pane_spawns_takes_input_and_survives_a_reconnect() { size: win(), shell: Some(plain_shell()), owner: None, + workspace: None, } .encode(&mut sock) .unwrap(); @@ -176,6 +177,7 @@ fn a_routed_kill_reaches_the_pane_it_names() { size: win(), shell: Some(plain_shell()), owner: None, + workspace: None, } .encode(&mut sock) .unwrap(); diff --git a/src/terminal/remote.rs b/src/terminal/remote.rs index a5002af1..4c84f221 100644 --- a/src/terminal/remote.rs +++ b/src/terminal/remote.rs @@ -252,6 +252,7 @@ impl RemoteTerminal { size: win, shell, owner, + workspace: None, } .encode(&mut stream)?; let pane_id = match DaemonMsg::read(&mut stream)? {