starred/tty7
1
0
Fork 0
mirror of https://github.com/l0ng-ai/tty7.git synced 2026-09-22 00:02:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(core): pane observers, TTY7_* context env, and control aggregate queries

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv
This commit is contained in:
thomas
2026-07-31 09:44:22 +08:00
co-authored by Claude Fable 5
parent 66b238ee8f
commit 753d1dceee
8 changed files with 888 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tty7-core/src/daemon/control.rs
+83 -1
View File
@@ -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<Instant> = 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<crate::core::cli_agent::CLIAgent>,
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<crate::core::machine::Workspace>),
TabTree(Box<Tab>),
Panes(Vec<u64>),
AgentStates(Vec<PaneAgentState>),
Routes(Vec<RouteInfo>),
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(),
crates/tty7-core/src/daemon/pane.rs
+337 -46
View File
@@ -112,13 +112,15 @@ struct SpawnConfig {
}
fn build_spawn_config(
pane: u64,
cwd: Option<PathBuf>,
shell: Option<ShellSpec>,
workspace: Option<&str>,
) -> anyhow::Result<SpawnConfig> {
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<RemoteContext> {
fn build_shell_command(
configured: Option<ChosenShell>,
initial_cwd: &Option<PathBuf>,
pane: u64,
workspace: Option<&str>,
) -> anyhow::Result<(CommandBuilder, Option<PathBuf>)> {
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<String> {
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<String> {
crate::host::server::control_socket_path()
.ok()
.map(|p| p.display().to_string())
}
#[cfg(windows)]
fn control_socket_env() -> Option<String> {
crate::host::server::control_endpoint_path()
.ok()
.map(|p| p.display().to_string())
}
#[cfg(not(any(unix, windows)))]
fn control_socket_env() -> Option<String> {
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<String, String>,
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<PathBuf>) {
fn apply_common_command_setup(
cmd: &mut CommandBuilder,
initial_cwd: &Option<PathBuf>,
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<Sender<DaemonMsg>>,
subscriber_epoch: u64,
observers: Vec<(u64, Sender<DaemonMsg>)>,
observer_seq: u64,
cwd: Option<PathBuf>,
shell: ShellState,
remote: Option<RemoteContext>,
@@ -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<ShellSpec>,
owner: Option<String>,
workspace: Option<String>,
on_dead: impl FnOnce() + Send + 'static,
) -> anyhow::Result<Arc<Self>> {
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<DaemonMsg>) -> 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<crate::daemon::control::PaneAgentState> {
agent_state_snapshot(&self.state.lock().unwrap())
}
pub fn gate(&self) -> Arc<OutputGate> {
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<DaemonMsg>) -> u64 {
st.subscriber_epoch += 1;
st.ring.replay(&subscriber);
fn replay_state(st: &PaneState, subscriber: &Sender<DaemonMsg>) {
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<DaemonMsg>) -> u64 {
if !st.alive {
let _ = subscriber.send(DaemonMsg::Exited { code: None });
}
}
fn attach_subscriber(st: &mut PaneState, subscriber: Sender<DaemonMsg>) -> u64 {
st.subscriber_epoch += 1;
replay_state(st, &subscriber);
st.subscriber = Some(subscriber);
st.subscriber_epoch
}
fn observe_subscriber(st: &mut PaneState, observer: Sender<DaemonMsg>) -> 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<crate::daemon::control::PaneAgentState> {
st.agent_session
.clone()
.map(|state| crate::daemon::control::PaneAgentState {
pane_id: st.id,
agent: st.agent,
state,
})
}
fn observed_facts(st: &PaneState) -> (Option<String>, Option<crate::core::machine::AgentFacts>) {
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<PathBuf>) {
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<RemoteContext>) {
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<Vec<String>>) {
&& 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<DaemonMsg>) -> Vec<DaemonMsg> {
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]
crates/tty7-core/src/daemon/protocol.rs
+82 -3
View File
@@ -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<ShellSpec>,
owner: Option<String>,
workspace: Option<String>,
},
Attach {
pane_id: u64,
size: WinSize,
},
Observe {
pane_id: u64,
size: WinSize,
},
Input(Vec<u8>),
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<ShellSpec>,
#[serde(default)]
owner: Option<String>,
#[serde(default)]
workspace: Option<String>,
}
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"
crates/tty7-core/src/daemon/server.rs
+154 -10
View File
@@ -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<crate::daemon::control::PaneAgentState> {
let panes: Vec<Arc<DaemonPane>> = 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<Registry>) {
@@ -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<Registry>) -> 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<Registry>) -> 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<Registry>) -> 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<Registry>) -> 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<DaemonPane>,
mut read_stream: Stream,
write_stream: Stream,
) -> anyhow::Result<()> {
let (tx, rx) = mpsc::channel::<DaemonMsg>();
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<R: std::io::Read>(read_stream: &mut R, refusals: &mpsc::Sender<DaemonMsg>) {
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<DaemonPane>,
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::<u8>::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();
crates/tty7-core/src/daemon/ssh/mod.rs
+58
View File
@@ -374,6 +374,26 @@ impl SshManager {
self.conns.lock().unwrap().remove(key);
}
pub fn routes(&self) -> Vec<crate::daemon::control::RouteInfo> {
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<SshConnection>) -> Option<String> {
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();
crates/tty7-core/src/host/server.rs
+171 -1
View File
@@ -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<PaneAgentState>;
}
#[derive(Clone, Default)]
pub struct Services {
pub machine: Option<Arc<MachineStore>>,
pub attachments: Arc<AttachRegistry>,
pub panes: Option<Arc<dyn PaneDirectory>>,
}
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<PathBuf> {
v.iter().map(PathBuf::from).collect()
}
@@ -745,6 +799,7 @@ struct Conn {
machine: Option<Arc<MachineStore>>,
machine_origin: Option<machine::SubscriberId>,
attachments: Arc<AttachRegistry>,
panes: Option<Arc<dyn PaneDirectory>>,
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<PaneAgentState> {
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::*;
crates/tty7-server/tests/routed_pane.rs
+2
View File
@@ -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();
src/terminal/remote.rs
+1
View File
@@ -252,6 +252,7 @@ impl RemoteTerminal {
size: win,
shell,
owner,
workspace: None,
}
.encode(&mut stream)?;
let pane_id = match DaemonMsg::read(&mut stream)? {