diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/api_ping.rs | 44 | ||||
| -rw-r--r-- | tests/auto_detect.rs | 706 | ||||
| -rw-r--r-- | tests/cli_wrapper.rs | 346 | ||||
| -rw-r--r-- | tests/client_mode.rs | 1221 | ||||
| -rw-r--r-- | tests/cross_area.rs | 1103 | ||||
| -rw-r--r-- | tests/detach_reattach.rs | 955 | ||||
| -rw-r--r-- | tests/multi_client.rs | 1019 | ||||
| -rw-r--r-- | tests/server_headless.rs | 713 | ||||
| -rw-r--r-- | tests/support/mod.rs | 418 |
9 files changed, 6462 insertions, 63 deletions
diff --git a/tests/api_ping.rs b/tests/api_ping.rs index acd7f1b8..b8de04f3 100644 --- a/tests/api_ping.rs +++ b/tests/api_ping.rs @@ -1,3 +1,5 @@ +mod support; + use std::fs; use std::io::{Read, Write}; use std::os::unix::net::UnixStream; @@ -7,6 +9,10 @@ use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; fn unique_test_dir() -> PathBuf { let nanos = SystemTime::now() @@ -21,24 +27,31 @@ struct SpawnedHerdr { child: Box<dyn Child + Send + Sync>, } -fn cleanup_spawned_herdr(mut spawned: SpawnedHerdr, base: PathBuf) { - let pid = spawned.child.process_id(); - let _ = spawned.child.kill(); - - if let Some(pid) = pid { - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - let mut status = 0; - let result = unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; - if result == pid as libc::pid_t || result == -1 { - break; +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); } - thread::sleep(Duration::from_millis(20)); + + unregister_spawned_herdr_pid(Some(pid)); } } +} +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { drop(spawned); - let _ = fs::remove_dir_all(base); + cleanup_test_base(&base); } fn test_lock() -> MutexGuard<'static, ()> { @@ -71,6 +84,7 @@ fn spawn_herdr_with_path( ) -> SpawnedHerdr { fs::create_dir_all(config_home.join("herdr")).unwrap(); fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); fs::write( config_home.join("herdr/config.toml"), "onboarding = false\n", @@ -87,10 +101,11 @@ fn spawn_herdr_with_path( .unwrap(); let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); - cmd.arg("--no-session"); + cmd.arg("server"); cmd.env("XDG_CONFIG_HOME", config_home); cmd.env("XDG_RUNTIME_DIR", runtime_dir); cmd.env("HERDR_SOCKET_PATH", socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); cmd.env("SHELL", "/bin/sh"); cmd.env_remove("HERDR_ENV"); if let Some(path) = path_override { @@ -98,6 +113,7 @@ fn spawn_herdr_with_path( } let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); SpawnedHerdr { _master: pair.master, diff --git a/tests/auto_detect.rs b/tests/auto_detect.rs new file mode 100644 index 00000000..4c92d62c --- /dev/null +++ b/tests/auto_detect.rs @@ -0,0 +1,706 @@ +//! Integration tests for auto-detect launch behavior. + +#![cfg(not(target_os = "macos"))] + +mod support; + +use std::fs; +use std::io::{BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use serde_json::Value; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-autodetect-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn wait_for_socket(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn spawn_server( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, + _client_socket_path: &Path, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +/// Spawn `herdr` (no subcommand) — the auto-detect launch path. +fn spawn_herdr_auto( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, + _client_socket_path: &Path, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + // No subcommand, no --no-session → auto-detect launch + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +/// Spawn `herdr --no-session` — the monolithic escape hatch. +fn spawn_herdr_no_session( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("--no-session"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn ping_socket(socket_path: &Path) -> String { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + + let request = r#"{"id":"1","method":"ping","params":{}}"#; + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + response.trim().to_string() +} + +fn run_cli(socket_path: &Path, args: &[&str]) -> std::process::Output { + let mut command = Command::new(env!("CARGO_BIN_EXE_herdr")); + command.args(args); + command.env("HERDR_SOCKET_PATH", socket_path); + command.output().unwrap() +} + +fn process_exists(pid: u32) -> bool { + let result = unsafe { libc::kill(pid as i32, 0) }; + if result == 0 { + true + } else { + std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } +} + +fn read_json_line(stream: UnixStream) -> Value { + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + serde_json::from_str(&response).unwrap() +} + +fn wait_for_pid_exit(pid: u32, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if !process_exists(pid) { + return true; + } + thread::sleep(Duration::from_millis(20)); + } + false +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// Running `herdr` with no server present starts a server +/// and attaches as client. +#[test] +fn auto_detect_no_server_spawns_server_and_attaches() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Ensure no server is running initially. + assert!( + !api_socket.exists(), + "api socket should not exist initially" + ); + assert!( + !client_socket.exists(), + "client socket should not exist initially" + ); + + // Run `herdr` (no subcommand) — should auto-detect, spawn server, attach as client. + let herdr = spawn_herdr_auto(&config_home, &runtime_dir, &api_socket, &client_socket); + + // Wait for both sockets to appear (server was spawned). + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Verify the API socket responds to ping (server is running). + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API socket should respond to ping: {response}" + ); + + // Verify the client socket accepts connections (server is listening on it). + let _stream = UnixStream::connect(&client_socket) + .expect("should connect to client socket (server is listening)"); + + // Verify the client process is running. + let client_pid = herdr.child.process_id().expect("client should have PID"); + assert!( + process_exists(client_pid), + "client process should be running" + ); + + cleanup_spawned_herdr(herdr, base); +} + +/// Running `herdr` with a server already running attaches +/// as client directly (no second server). +#[test] +fn auto_detect_server_running_attaches_directly() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Start a server explicitly. + let server = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let server_pid = server.child.process_id().expect("server should have PID"); + + // Verify server is running. + assert!(process_exists(server_pid), "server should be running"); + + // Run `herdr` (no subcommand) — should detect the running server and attach. + let client = spawn_herdr_auto(&config_home, &runtime_dir, &api_socket, &client_socket); + + // Wait a moment for the client to attach. + thread::sleep(Duration::from_millis(500)); + + // Verify the client is running. + let client_pid = client.child.process_id().expect("client should have PID"); + assert!( + process_exists(client_pid), + "client process should be running" + ); + + // Verify the server is still the same one (no second server spawned). + assert!( + process_exists(server_pid), + "original server should still be running" + ); + + // Verify API still responds. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should still respond to ping: {response}" + ); + + cleanup_spawned_herdr(client, PathBuf::from("/nonexistent")); + cleanup_spawned_herdr(server, base); +} + +/// Socket path resolution is consistent between server and client. +/// Both derive the client socket from the `HERDR_SOCKET_PATH` override, +/// so overriding the API socket keeps both endpoints aligned. +#[test] +fn auto_detect_socket_path_consistency() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Run `herdr` with custom socket paths. + let herdr = spawn_herdr_auto(&config_home, &runtime_dir, &api_socket, &client_socket); + + // Wait for both sockets to appear at the custom paths. + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Verify sockets exist at the specified paths. + assert!( + api_socket.exists(), + "API socket should exist at custom path" + ); + assert!( + client_socket.exists(), + "client socket should exist at custom path" + ); + + // Verify API responds (server is using the custom API socket path). + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should respond at custom path: {response}" + ); + + // Verify client socket accepts connections (server is using the custom + // client socket path). + let _stream = UnixStream::connect(&client_socket) + .expect("should connect to client socket at custom path"); + + cleanup_spawned_herdr(herdr, base); +} + +/// `herdr --no-session` bypasses server/client and runs +/// monolithically. No server process is spawned. No client socket is created. +#[test] +fn no_session_flag_runs_monolithically() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Run `herdr --no-session` — monolithic mode, no server/client. + let herdr = spawn_herdr_no_session(&config_home, &runtime_dir, &api_socket); + + // Wait for the API socket (monolithic mode creates it). + wait_for_socket(&api_socket, Duration::from_secs(10)); + + // Verify the API socket exists and responds. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "monolithic API should respond: {response}" + ); + + // Verify NO client socket was created — this is the key distinction + // between monolithic mode and server/client mode. + assert!( + !client_socket.exists(), + "no client socket should exist in monolithic mode" + ); + + // Verify the API socket is served by the monolithic process itself, + // not by a separate server. We can check this by verifying the client + // PID matches what would be serving the socket — in monolithic mode, + // there is only one herdr process. + let client_pid = herdr.child.process_id().expect("should have PID"); + assert!( + process_exists(client_pid), + "monolithic process should be running" + ); + + cleanup_spawned_herdr(herdr, base); +} + +/// CLI subcommands work through the server's JSON API socket. +#[test] +fn cli_subcommands_work_through_server() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Start a server. + let server = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Test `herdr workspace list` through the server's API socket. + let output = run_cli(&api_socket, &["workspace", "list"]); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success(), + "workspace list should succeed: stderr={}", + String::from_utf8_lossy(&output.stderr) + ); + // The response should be valid JSON with a "result" field. + assert!( + stdout.contains("result"), + "workspace list output should contain 'result': {stdout}" + ); + + // Test `herdr pane list` through the server's API socket. + let output = run_cli(&api_socket, &["pane", "list"]); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success(), + "pane list should succeed: stderr={}", + String::from_utf8_lossy(&output.stderr) + ); + assert!( + stdout.contains("result"), + "pane list output should contain 'result': {stdout}" + ); + + cleanup_spawned_herdr(server, base); +} + +/// Verify that the server spawned by auto-detect +/// persists after the client exits, and a new `herdr` can reattach. +#[test] +fn auto_detect_server_persists_and_reattaches() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Run `herdr` — auto-detect spawns server + attaches client. + let mut client1 = spawn_herdr_auto(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Verify API responds. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should respond before client exit: {response}" + ); + + // Kill the first client. + let client1_pid = client1.child.process_id().expect("client1 should have PID"); + let _ = client1.child.kill(); + let _ = wait_for_pid_exit(client1_pid, Duration::from_secs(2)); + drop(client1); + + // Wait a moment for the server to process the disconnect. + thread::sleep(Duration::from_millis(500)); + + // Verify server is still running after client exit — the API should + // still respond because the server is a separate daemon process. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should still respond after client exit (server persists): {response}" + ); + + // The client socket should still exist (server is still listening). + assert!( + client_socket.exists(), + "client socket should still exist after client exit" + ); + + // Run `herdr` again — should detect the running server and reattach. + let client2 = spawn_herdr_auto(&config_home, &runtime_dir, &api_socket, &client_socket); + thread::sleep(Duration::from_millis(500)); + + // Verify the new client is running. + let client2_pid = client2.child.process_id().expect("client2 should have PID"); + assert!( + process_exists(client2_pid), + "second client should be running" + ); + + // Verify API still responds (same server). + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should still respond after reattach: {response}" + ); + + cleanup_spawned_herdr(client2, PathBuf::from("/nonexistent")); + cleanup_test_base(&base); +} + +/// Verify that the default API and client +/// sockets live in the app config directory when no env override is set. +#[test] +fn auto_detect_default_socket_path_from_config_dir() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + + // Don't set HERDR_SOCKET_PATH or HERDR_CLIENT_SOCKET_PATH. + // The default paths should come from the app config directory, not XDG_RUNTIME_DIR. + let app_dir_name = if cfg!(debug_assertions) { + "herdr-dev" + } else { + "herdr" + }; + let api_socket = config_home.join(app_dir_name).join("herdr.sock"); + let client_socket = config_home.join(app_dir_name).join("herdr-client.sock"); + + // Spawn server with XDG_RUNTIME_DIR set to a different directory to prove it is ignored. + fs::create_dir_all(config_home.join(app_dir_name)).unwrap(); + fs::create_dir_all(&runtime_dir).unwrap(); + register_runtime_dir(&runtime_dir); + fs::write( + config_home.join(app_dir_name).join("config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", &config_home); + cmd.env("XDG_RUNTIME_DIR", &runtime_dir); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + // Explicitly remove socket overrides to test default path resolution. + cmd.env_remove("HERDR_SOCKET_PATH"); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + let server = SpawnedHerdr { + _master: pair.master, + child, + }; + + // Wait for sockets to appear at the default config-dir paths. + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Verify both sockets exist. + assert!(api_socket.exists(), "API socket should exist in config dir"); + assert!( + client_socket.exists(), + "client socket should exist in config dir" + ); + + // Verify API responds. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should respond at config-dir path: {response}" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn auto_detect_respects_nested_guard_before_auto_attach() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let baseline = read_json_line({ + let mut stream = UnixStream::connect(&api_socket).unwrap(); + writeln!( + stream, + r#"{{"id":"ws_before","method":"workspace.list","params":{{}}}}"# + ) + .unwrap(); + stream + }); + let baseline_count = baseline["result"]["workspaces"] + .as_array() + .map(|workspaces| workspaces.len()) + .unwrap_or(0); + + let output = Command::new(env!("CARGO_BIN_EXE_herdr")) + .env("XDG_CONFIG_HOME", &config_home) + .env("XDG_RUNTIME_DIR", &runtime_dir) + .env("HERDR_SOCKET_PATH", &api_socket) + .env_remove("HERDR_CLIENT_SOCKET_PATH") + .env("HERDR_ENV", "1") + .output() + .unwrap(); + + assert!( + !output.status.success(), + "nested launch should fail before auto-attach" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("nested herdr is disabled by default"), + "stderr should mention nested-launch guard: {stderr}" + ); + + let after = read_json_line({ + let mut stream = UnixStream::connect(&api_socket).unwrap(); + writeln!( + stream, + r#"{{"id":"ws_after","method":"workspace.list","params":{{}}}}"# + ) + .unwrap(); + stream + }); + let after_count = after["result"]["workspaces"] + .as_array() + .map(|workspaces| workspaces.len()) + .unwrap_or(0); + assert_eq!( + after_count, baseline_count, + "nested launch should not auto-attach or mutate server state" + ); + + cleanup_spawned_herdr(server, base); +} diff --git a/tests/cli_wrapper.rs b/tests/cli_wrapper.rs index 8c8574d3..92684bff 100644 --- a/tests/cli_wrapper.rs +++ b/tests/cli_wrapper.rs @@ -1,5 +1,7 @@ #![cfg(not(target_os = "macos"))] +mod support; + use std::fs; use std::io::{BufRead, BufReader, Write}; use std::os::unix::net::{UnixListener, UnixStream}; @@ -9,6 +11,10 @@ use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; fn unique_test_dir() -> PathBuf { let nanos = SystemTime::now() @@ -23,24 +29,31 @@ struct SpawnedHerdr { child: Box<dyn Child + Send + Sync>, } -fn cleanup_spawned_herdr(mut spawned: SpawnedHerdr, base: PathBuf) { - let pid = spawned.child.process_id(); - let _ = spawned.child.kill(); - - if let Some(pid) = pid { - let deadline = Instant::now() + Duration::from_secs(2); - while Instant::now() < deadline { - let mut status = 0; - let result = unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; - if result == pid as libc::pid_t || result == -1 { - break; +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); } - thread::sleep(Duration::from_millis(20)); + + unregister_spawned_herdr_pid(Some(pid)); } } +} +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { drop(spawned); - let _ = fs::remove_dir_all(base); + cleanup_test_base(&base); } fn wait_for_socket(path: &Path, timeout: Duration) { @@ -66,6 +79,7 @@ fn spawn_herdr_with_path( ) -> SpawnedHerdr { fs::create_dir_all(config_home.join("herdr")).unwrap(); fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); fs::write( config_home.join("herdr/config.toml"), "onboarding = false\n", @@ -82,10 +96,11 @@ fn spawn_herdr_with_path( .unwrap(); let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); - cmd.arg("--no-session"); + cmd.arg("server"); cmd.env("XDG_CONFIG_HOME", config_home); cmd.env("XDG_RUNTIME_DIR", runtime_dir); cmd.env("HERDR_SOCKET_PATH", socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); cmd.env("SHELL", "/bin/sh"); cmd.env_remove("HERDR_ENV"); if let Some(path) = path_override { @@ -93,6 +108,7 @@ fn spawn_herdr_with_path( } let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); SpawnedHerdr { _master: pair.master, child, @@ -106,6 +122,28 @@ fn run_cli(socket_path: &Path, args: &[&str]) -> std::process::Output { command.output().unwrap() } +fn run_cli_json(socket_path: &Path, args: &[&str]) -> serde_json::Value { + let output = run_cli(socket_path, args); + assert!( + output.status.success(), + "command failed: herdr {}\nstatus: {:?}\nstderr: {}\nstdout: {}", + args.join(" "), + output.status.code(), + String::from_utf8_lossy(&output.stderr), + String::from_utf8_lossy(&output.stdout) + ); + + serde_json::from_slice(&output.stdout).unwrap_or_else(|err| { + panic!( + "failed to parse JSON response for `herdr {}`: {}\nstdout: {}\nstderr: {}", + args.join(" "), + err, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ) + }) +} + fn process_exists(pid: u32) -> bool { let result = unsafe { libc::kill(pid as i32, 0) }; if result == 0 { @@ -126,6 +164,118 @@ fn wait_for_pid_exit(pid: u32, timeout: Duration) -> bool { !process_exists(pid) } +fn wait_for_pid_file(pid_file: &Path, timeout: Duration) -> Result<u32, String> { + const STABLE_PID_CONTENT_WINDOW: Duration = Duration::from_millis(250); + + let deadline = Instant::now() + timeout; + let mut last_contents = String::new(); + let mut stable_candidate: Option<(String, u32, Instant)> = None; + + while Instant::now() < deadline { + if let Ok(contents) = fs::read_to_string(pid_file) { + let trimmed = contents.trim().to_string(); + last_contents = contents; + + if let Ok(pid) = trimmed.parse::<u32>() { + match &stable_candidate { + Some((candidate_text, candidate_pid, stable_since)) + if candidate_text == &trimmed && *candidate_pid == pid => + { + if stable_since.elapsed() >= STABLE_PID_CONTENT_WINDOW { + return Ok(pid); + } + } + _ => { + stable_candidate = Some((trimmed, pid, Instant::now())); + } + } + } else { + stable_candidate = None; + } + } + + thread::sleep(Duration::from_millis(25)); + } + + Err(format!( + "pid file {} did not contain stable parseable pid before timeout; last contents={:?}", + pid_file.display(), + last_contents + )) +} + +#[test] +fn wait_for_pid_file_retries_until_pid_is_written() { + let base = unique_test_dir(); + fs::create_dir_all(&base).unwrap(); + let pid_file = base.join("delayed.pid"); + fs::write(&pid_file, "").unwrap(); + + let writer = thread::spawn({ + let pid_file = pid_file.clone(); + move || { + thread::sleep(Duration::from_millis(100)); + fs::write(pid_file, "424242\n").unwrap(); + } + }); + + let pid = wait_for_pid_file(&pid_file, Duration::from_secs(2)).unwrap(); + assert_eq!(pid, 424242); + + writer.join().unwrap(); + cleanup_test_base(&base); +} + +#[test] +fn wait_for_pid_file_errors_when_file_never_contains_pid() { + let base = unique_test_dir(); + fs::create_dir_all(&base).unwrap(); + let pid_file = base.join("empty.pid"); + fs::write(&pid_file, "").unwrap(); + + let err = wait_for_pid_file(&pid_file, Duration::from_millis(150)).unwrap_err(); + assert!( + err.contains("did not contain stable parseable pid"), + "unexpected error: {err}" + ); + + cleanup_test_base(&base); +} + +#[test] +fn wait_for_pid_file_rejects_partial_write_race_until_stable_contents() { + let base = unique_test_dir(); + fs::create_dir_all(&base).unwrap(); + let pid_file = base.join("partial-race.pid"); + fs::write(&pid_file, "").unwrap(); + + let writer = thread::spawn({ + let pid_file = pid_file.clone(); + move || { + thread::sleep(Duration::from_millis(40)); + fs::write(&pid_file, "12").unwrap(); + thread::sleep(Duration::from_millis(40)); + fs::write(&pid_file, "123").unwrap(); + thread::sleep(Duration::from_millis(40)); + fs::write(&pid_file, "1234").unwrap(); + thread::sleep(Duration::from_millis(200)); + fs::write(&pid_file, "424242\n").unwrap(); + } + }); + + let start = Instant::now(); + let pid = wait_for_pid_file(&pid_file, Duration::from_secs(2)).unwrap(); + assert_eq!(pid, 424242); + assert!( + start.elapsed() >= Duration::from_millis(300), + "helper should wait for stable complete contents, elapsed={:?}", + start.elapsed() + ); + + writer.join().unwrap(); + cleanup_test_base(&base); +} + fn send_request(socket_path: &Path, json: &str) -> serde_json::Value { let mut stream = UnixStream::connect(socket_path).unwrap(); stream.write_all(json.as_bytes()).unwrap(); @@ -205,7 +355,106 @@ fn pane_run_sends_one_send_input_request_with_enter_key() { second_line ); - fs::remove_dir_all(base).unwrap(); + cleanup_test_base(&base); +} + +#[test] +fn integration_commands_honor_socket_override_when_server_is_missing() { + let base = unique_test_dir(); + let home_dir = base.join("home"); + let extensions_dir = home_dir.join(".pi/agent/extensions"); + fs::create_dir_all(&extensions_dir).unwrap(); + + let runtime_dir = base.join("runtime"); + fs::create_dir_all(&runtime_dir).unwrap(); + register_runtime_dir(&runtime_dir); + let missing_socket = runtime_dir.join("missing.sock"); + + let expected_extension = extensions_dir.join("herdr-agent-state.ts"); + assert!( + !expected_extension.exists(), + "test setup should start without extension file" + ); + + let workspace_list = Command::new(env!("CARGO_BIN_EXE_herdr")) + .args(["workspace", "list"]) + .env("HERDR_SOCKET_PATH", &missing_socket) + .env("HOME", &home_dir) + .output() + .unwrap(); + assert_eq!(workspace_list.status.code(), Some(1)); + + let integration_install = Command::new(env!("CARGO_BIN_EXE_herdr")) + .args(["integration", "install", "pi"]) + .env("HERDR_SOCKET_PATH", &missing_socket) + .env("HOME", &home_dir) + .output() + .unwrap(); + assert_eq!(integration_install.status.code(), Some(1)); + assert!( + !expected_extension.exists(), + "integration install should not run local install logic when socket is missing" + ); + + let integration_uninstall = Command::new(env!("CARGO_BIN_EXE_herdr")) + .args(["integration", "uninstall", "pi"]) + .env("HERDR_SOCKET_PATH", &missing_socket) + .env("HOME", &home_dir) + .output() + .unwrap(); + assert_eq!(integration_uninstall.status.code(), Some(1)); + assert!( + !expected_extension.exists(), + "integration uninstall should also be socket-backed when socket is missing" + ); + + cleanup_test_base(&base); +} + +#[test] +fn server_stop_command_shuts_down_running_server() { + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let socket_path = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut herdr = spawn_herdr(&config_home, &runtime_dir, &socket_path); + wait_for_socket(&socket_path, Duration::from_secs(5)); + wait_for_socket(&client_socket, Duration::from_secs(5)); + + let stopped = run_cli(&socket_path, &["server", "stop"]); + assert!( + stopped.status.success(), + "stderr: {}", + String::from_utf8_lossy(&stopped.stderr) + ); + assert!( + stopped.stdout.is_empty(), + "server stop should not print stdout: {}", + String::from_utf8_lossy(&stopped.stdout) + ); + + let pid = herdr.child.process_id(); + let exit_status = herdr.child.wait().unwrap(); + unregister_spawned_herdr_pid(pid); + assert!(exit_status.success(), "server stop should exit cleanly"); + + let deadline = Instant::now() + Duration::from_secs(3); + while Instant::now() < deadline && (socket_path.exists() || client_socket.exists()) { + thread::sleep(Duration::from_millis(25)); + } + + assert!( + !socket_path.exists() || UnixStream::connect(&socket_path).is_err(), + "api socket should be removed or stale after server stop" + ); + assert!( + !client_socket.exists() || UnixStream::connect(&client_socket).is_err(), + "client socket should be removed or stale after server stop" + ); + + cleanup_spawned_herdr(herdr, base); } #[test] @@ -636,11 +885,9 @@ fn closing_pane_terminates_processes_inside_it() { } assert!(pid_file.exists(), "pid file was not created"); - let pid: u32 = fs::read_to_string(&pid_file) - .unwrap() - .trim() - .parse() - .unwrap(); + let pid = wait_for_pid_file(&pid_file, Duration::from_secs(3)).unwrap_or_else(|err| { + panic!("failed to read pane child pid: {err}"); + }); assert!(process_exists(pid), "child process was not running"); let closed = run_cli(&socket_path, &["pane", "close", pane_id]); @@ -691,11 +938,9 @@ fn closing_workspace_terminates_processes_inside_it() { } assert!(pid_file.exists(), "pid file was not created"); - let pid: u32 = fs::read_to_string(&pid_file) - .unwrap() - .trim() - .parse() - .unwrap(); + let pid = wait_for_pid_file(&pid_file, Duration::from_secs(3)).unwrap_or_else(|err| { + panic!("failed to read pane child pid: {err}"); + }); assert!(process_exists(pid), "child process was not running"); let closed = run_cli(&socket_path, &["workspace", "close", "1"]); @@ -722,42 +967,37 @@ fn workspace_ids_are_stable_and_pane_numbers_stay_compact() { let herdr = spawn_herdr(&config_home, &runtime_dir, &socket_path); wait_for_socket(&socket_path, Duration::from_secs(5)); - let ws1 = run_cli( + let ws1_json = run_cli_json( &socket_path, &["workspace", "create", "--cwd", base.to_str().unwrap()], ); - assert!(ws1.status.success()); - let ws1_json: serde_json::Value = serde_json::from_slice(&ws1.stdout).unwrap(); let ws1_id = ws1_json["result"]["workspace"]["workspace_id"] .as_str() .unwrap() .to_string(); - let split_12 = run_cli( + let split_12_json = run_cli_json( &socket_path, &["pane", "split", "1-1", "--direction", "right", "--no-focus"], ); - let split_12_json: serde_json::Value = serde_json::from_slice(&split_12.stdout).unwrap(); assert_eq!( split_12_json["result"]["pane"]["pane_id"], format!("{ws1_id}-2") ); - let split_13 = run_cli( + let split_13_json = run_cli_json( &socket_path, &["pane", "split", "1-1", "--direction", "down", "--no-focus"], ); - let split_13_json: serde_json::Value = serde_json::from_slice(&split_13.stdout).unwrap(); assert_eq!( split_13_json["result"]["pane"]["pane_id"], format!("{ws1_id}-3") ); - let ws2 = run_cli( + let ws2_json = run_cli_json( &socket_path, &["workspace", "create", "--cwd", "/tmp", "--no-focus"], ); - let ws2_json: serde_json::Value = serde_json::from_slice(&ws2.stdout).unwrap(); let ws2_id = ws2_json["result"]["workspace"]["workspace_id"] .as_str() .unwrap() @@ -765,22 +1005,25 @@ fn workspace_ids_are_stable_and_pane_numbers_stay_compact() { assert_ne!(ws2_id, ws1_id); let ws2_focus = run_cli(&socket_path, &["workspace", "focus", &ws2_id]); - assert!(ws2_focus.status.success()); - let ws2_split = run_cli( + assert!( + ws2_focus.status.success(), + "stderr: {}", + String::from_utf8_lossy(&ws2_focus.stderr) + ); + + let ws2_split_json = run_cli_json( &socket_path, &["pane", "split", "2-1", "--direction", "right", "--no-focus"], ); - let ws2_split_json: serde_json::Value = serde_json::from_slice(&ws2_split.stdout).unwrap(); assert_eq!( ws2_split_json["result"]["pane"]["pane_id"], format!("{ws2_id}-2") ); - let ws3 = run_cli( + let ws3_json = run_cli_json( &socket_path, &["workspace", "create", "--cwd", "/", "--no-focus"], ); - let ws3_json: serde_json::Value = serde_json::from_slice(&ws3.stdout).unwrap(); let ws3_id = ws3_json["result"]["workspace"]["workspace_id"] .as_str() .unwrap() @@ -789,10 +1032,13 @@ fn workspace_ids_are_stable_and_pane_numbers_stay_compact() { assert_ne!(ws3_id, ws2_id); let close_ws2 = run_cli(&socket_path, &["workspace", "close", &ws2_id]); - assert!(close_ws2.status.success()); + assert!( + close_ws2.status.success(), + "stderr: {}", + String::from_utf8_lossy(&close_ws2.stderr) + ); - let workspaces = run_cli(&socket_path, &["workspace", "list"]); - let workspaces_json: serde_json::Value = serde_json::from_slice(&workspaces.stdout).unwrap(); + let workspaces_json = run_cli_json(&socket_path, &["workspace", "list"]); let ids: Vec<String> = workspaces_json["result"]["workspaces"] .as_array() .unwrap() @@ -801,11 +1047,10 @@ fn workspace_ids_are_stable_and_pane_numbers_stay_compact() { .collect(); assert_eq!(ids, vec![ws1_id.clone(), ws3_id.clone()]); - let new_ws = run_cli( + let new_ws_json = run_cli_json( &socket_path, &["workspace", "create", "--cwd", "/var/tmp", "--no-focus"], ); - let new_ws_json: serde_json::Value = serde_json::from_slice(&new_ws.stdout).unwrap(); let new_ws_id = new_ws_json["result"]["workspace"]["workspace_id"] .as_str() .unwrap() @@ -814,17 +1059,20 @@ fn workspace_ids_are_stable_and_pane_numbers_stay_compact() { assert_ne!(new_ws_id, ws2_id); assert_ne!(new_ws_id, ws3_id); - let ws3_panes = run_cli(&socket_path, &["pane", "list", "--workspace", &ws3_id]); - let ws3_panes_json: serde_json::Value = serde_json::from_slice(&ws3_panes.stdout).unwrap(); + let ws3_panes_json = run_cli_json(&socket_path, &["pane", "list", "--workspace", &ws3_id]); assert_eq!( ws3_panes_json["result"]["panes"][0]["pane_id"], format!("{ws3_id}-1") ); let close_middle = run_cli(&socket_path, &["pane", "close", &format!("{ws1_id}-2")]); - assert!(close_middle.status.success()); - let ws1_panes = run_cli(&socket_path, &["pane", "list", "--workspace", &ws1_id]); - let ws1_panes_json: serde_json::Value = serde_json::from_slice(&ws1_panes.stdout).unwrap(); + assert!( + close_middle.status.success(), + "stderr: {}", + String::from_utf8_lossy(&close_middle.stderr) + ); + + let ws1_panes_json = run_cli_json(&socket_path, &["pane", "list", "--workspace", &ws1_id]); let pane_ids: Vec<String> = ws1_panes_json["result"]["panes"] .as_array() .unwrap() diff --git a/tests/client_mode.rs b/tests/client_mode.rs new file mode 100644 index 00000000..882bcb1c --- /dev/null +++ b/tests/client_mode.rs @@ -0,0 +1,1221 @@ +//! Integration tests for thin client mode. + +mod support; + +use std::fs; +use std::io::{BufRead, BufReader, Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-client-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn wait_for_socket(path: &PathBuf, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn wait_for_file(path: &PathBuf, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("file did not appear at {}", path.display()); +} + +fn spawn_client_process( + config_home: &PathBuf, + runtime_dir: &PathBuf, + api_socket_path: &PathBuf, +) -> SpawnedHerdr { + register_runtime_dir(runtime_dir); + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("client"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn spawn_server( + config_home: &PathBuf, + runtime_dir: &PathBuf, + api_socket_path: &PathBuf, + _client_socket_path: &PathBuf, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn ping_socket(socket_path: &PathBuf) -> String { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + + let request = r#"{"id":"1","method":"ping","params":{}}"#; + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + response.trim().to_string() +} + +// --------------------------------------------------------------------------- +// Bincode v2 varint encoding helpers (same as server_headless.rs tests) +// --------------------------------------------------------------------------- + +/// Encode a varint u32 value according to bincode v2 VarintEncoding. +fn encode_varint_u32(v: u32) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else if v < 65536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode a varint u16 value. +fn encode_varint_u16(v: u16) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode an enum variant with its fields. +fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> { + let mut buf = encode_varint_u32(variant_idx); + for field in fields { + buf.extend_from_slice(field); + } + buf +} + +/// Frame a message with u32LE length prefix. +fn frame_message(payload: &[u8]) -> Vec<u8> { + let len = payload.len() as u32; + let mut framed = len.to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +/// Decode a varint u32 from a byte slice at the given offset. +fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first_byte}")), + } +} + +/// Sends a Hello message over the client socket and reads the Welcome response. +fn client_handshake( + stream: &mut UnixStream, + version: u32, + cols: u16, + rows: u16, +) -> Result<(u32, Option<String>), String> { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + + // Encode Hello message using bincode v2 varint format. + // ClientMessage::Hello is variant 0. + let hello_payload = encode_varint_enum( + 0, + &[ + &encode_varint_u32(version), + &encode_varint_u16(cols), + &encode_varint_u16(rows), + ], + ); + let framed = frame_message(&hello_payload); + stream.write_all(&framed).map_err(|e| e.to_string())?; + stream.flush().map_err(|e| e.to_string())?; + + // Read the framed response. + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; + let len = u32::from_le_bytes(len_buf) as usize; + + if len > 2 * 1024 * 1024 { + return Err(format!("oversized response: {len}")); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).map_err(|e| e.to_string())?; + + // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option<String> } + decode_welcome(&payload) +} + +/// Decode a ServerMessage::Welcome from bincode v2 payload. +fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> { + let mut offset = 0; + + // Variant index (should be 0 for Welcome) + let (variant, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + if variant != 0 { + return Err(format!( + "expected Welcome (variant 0), got variant {variant}" + )); + } + + // version: u32 + let (version, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + + // error: Option<String> — discriminant is always 1 byte + if offset >= payload.len() { + return Err("payload too short for Option tag".into()); + } + let option_tag = payload[offset]; + offset += 1; + + let error = if option_tag == 1 { + // Some(String) — length as varint + UTF-8 bytes + let (str_len, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + let str_len = str_len as usize; + + if offset + str_len > payload.len() { + return Err("payload too short for string content".into()); + } + let s = String::from_utf8(payload[offset..offset + str_len].to_vec()) + .map_err(|e| e.to_string())?; + Some(s) + } else { + None + }; + + Ok((version, error)) +} + +/// Read a framed ServerMessage from the stream, returning the variant index +/// and the raw payload for further decoding. +fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> { + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .map_err(|e| e.to_string())?; + + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .map_err(|e| format!("read length prefix: {e}"))?; + let len = u32::from_le_bytes(len_buf) as usize; + + if len > 2 * 1024 * 1024 { + return Err(format!("oversized frame: {len} bytes")); + } + if len == 0 { + return Err("zero-length frame".into()); + } + + let mut payload = vec![0u8; len]; + stream + .read_exact(&mut payload) + .map_err(|e| format!("read payload: {e}"))?; + + // Decode the variant index. + let (variant, consumed) = decode_varint_u32(&payload, 0)?; + + // Return the variant index and the remaining payload (after the variant tag). + Ok((variant, payload[consumed..].to_vec())) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn client_connects_and_receives_frame() { + // Client connects to server and handshake completes. + // Client receives Frame messages. + // Server sends rendered frames to connected clients. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1, "server should report protocol version 1"); + assert!( + error.is_none(), + "handshake should not have error: {:?}", + error + ); + + // Give the server a moment to set up the writer thread and + // start the render cycle after learning our terminal size. + thread::sleep(Duration::from_millis(500)); + + // Read the next message from the server — should be a Frame (variant 1). + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + + let (variant, _payload) = + read_server_message(&mut stream).expect("should receive a message from server"); + + // ServerMessage::Frame is variant 1. + assert_eq!( + variant, 1, + "expected Frame message (variant 1), got variant {variant}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn client_input_forwarded_to_pane() { + // Stdin input is forwarded to server as ClientMessage::Input. + // Server routes client input to the correct PTY. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Send an Input message containing "echo hello\n". + // ClientMessage::Input is variant 1: { data: Vec<u8> } + let input_data = b"echo hello\n".to_vec(); + let input_payload = { + let mut buf = encode_varint_u32(1); // variant 1 = Input + // Encode the data as a bincode Vec<u8>: length (varint) + bytes + buf.extend_from_slice(&encode_varint_u32(input_data.len() as u32)); + buf.extend_from_slice(&input_data); + buf + }; + let framed = frame_message(&input_payload); + stream + .write_all(&framed) + .expect("should send Input message"); + stream.flush().expect("should flush"); + + // Give the server time to process the input and render. + thread::sleep(Duration::from_millis(500)); + + // Verify the server is still alive and responsive via API. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should still respond to ping after input: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn client_resize_sends_message() { + // Terminal resize triggers ClientMessage::Resize. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain the initial frame(s). + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + while let Ok(_) = read_server_message(&mut stream) {} + + // Send a Resize message: ClientMessage::Resize is variant 2: { cols: u16, rows: u16 } + let resize_payload = { + let mut buf = encode_varint_u32(2); // variant 2 = Resize + buf.extend_from_slice(&encode_varint_u16(120)); // cols + buf.extend_from_slice(&encode_varint_u16(40)); // rows + buf + }; + let framed = frame_message(&resize_payload); + stream + .write_all(&framed) + .expect("should send Resize message"); + stream.flush().expect("should flush"); + + // Give the server time to process and render at the new size. + thread::sleep(Duration::from_millis(500)); + + // Verify the server is still alive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should respond after resize: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_shutdown_sends_message_to_client() { + // ServerShutdown causes clean exit with informative message. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Send SIGINT so the server takes the graceful shutdown path and + // broadcasts ServerShutdown before exiting. + if let Some(pid) = spawned.child.process_id() { + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGINT); + } + } + + // The client should receive an explicit ServerShutdown message, or at + // minimum observe clean connection close if shutdown races with send. + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut saw_shutdown = false; + let mut saw_disconnect = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream) { + Ok((variant, _)) => { + if variant == 2 { + saw_shutdown = true; + break; + } + } + Err(_) => { + saw_disconnect = true; + break; + } + } + } + assert!( + saw_shutdown || saw_disconnect, + "client should observe ServerShutdown or disconnect during graceful shutdown" + ); + + // Wait for the server to exit after shutdown signal. + let _ = spawned.child.wait(); + + drop(spawned); + cleanup_test_base(&base); +} + +#[test] +fn server_unreachable_shows_clear_error() { + // when server is unreachable, the client exits quickly + // with an actionable connection-failed message. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(&runtime_dir).unwrap(); + register_runtime_dir(&runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let output = std::process::Command::new(env!("CARGO_BIN_EXE_herdr")) + .arg("client") + .env("XDG_CONFIG_HOME", &config_home) + .env("XDG_RUNTIME_DIR", &runtime_dir) + .env("HERDR_SOCKET_PATH", &api_socket) + .env_remove("HERDR_CLIENT_SOCKET_PATH") + .env_remove("HERDR_ENV") + .output() + .expect("client command should run"); + + assert!( + !output.status.success(), + "client should fail when no server is running" + ); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("failed to connect to server"), + "stderr should mention connection failure: {stderr}" + ); + assert!( + stderr.contains("Is herdr server running?"), + "stderr should include actionable guidance: {stderr}" + ); + assert!( + stderr.contains("Socket path:"), + "stderr should include attempted socket path: {stderr}" + ); + + cleanup_test_base(&base); +} + +#[test] +fn server_crash_after_attach_causes_lost_connection_error() { + // attach a real thin client connection, kill server unexpectedly, + // assert clean non-zero client exit plus lost-connection signal. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Attach a real thin client (client subcommand) through PTY so handshake and + // terminal setup paths are exercised. + let mut thin_client = spawn_client_process(&config_home, &runtime_dir, &api_socket); + + // Prove attached before kill by waiting for at least one frame message. + let mut thin_reader = thin_client + ._master + .try_clone_reader() + .expect("clone client PTY reader"); + let attached_before_kill = { + let deadline = Instant::now() + Duration::from_secs(8); + let mut buf = [0u8; 4096]; + let mut seen = false; + while Instant::now() < deadline { + match thin_reader.read(&mut buf) { + Ok(n) if n > 0 => { + let out = String::from_utf8_lossy(&buf[..n]); + if out.contains("\u{2500}") + || out.contains("workspace") + || out.contains("terminal") + || out.contains("pane") + { + seen = true; + break; + } + } + Ok(_) => thread::sleep(Duration::from_millis(30)), + Err(_) => thread::sleep(Duration::from_millis(30)), + } + } + seen + }; + assert!( + attached_before_kill, + "thin client must complete attach and receive frame before server crash" + ); + + // Kill server unexpectedly. + if let Some(pid) = spawned.child.process_id() { + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGKILL); + } + } + + // Client should exit non-zero after connection loss. + let mut crash_output = String::new(); + let exited = { + let deadline = Instant::now() + Duration::from_secs(12); + let mut exited = false; + while Instant::now() < deadline { + if thin_client.child.try_wait().ok().flatten().is_some() { + exited = true; + break; + } + // Keep draining client output so the process can progress to exit. + let mut buf = [0u8; 1024]; + if let Ok(n) = thin_reader.read(&mut buf) { + if n > 0 { + crash_output.push_str(&String::from_utf8_lossy(&buf[..n])); + } + } + thread::sleep(Duration::from_millis(50)); + } + exited + }; + assert!(exited, "thin client should exit after server SIGKILL"); + + let status = thin_client.child.wait().expect("wait thin client status"); + assert!( + !status.success(), + "thin client should exit non-zero after lost server connection" + ); + + // Drain trailing output and require the explicit user-visible lost-connection message. + let deadline = Instant::now() + Duration::from_secs(5); + let mut buf = [0u8; 2048]; + while Instant::now() < deadline { + match thin_reader.read(&mut buf) { + Ok(n) if n > 0 => crash_output.push_str(&String::from_utf8_lossy(&buf[..n])), + Ok(_) => break, + Err(_) => break, + } + thread::sleep(Duration::from_millis(30)); + } + + let crash_output_lc = crash_output.to_lowercase(); + assert!( + crash_output_lc.contains("lost connection to server"), + "thin client must emit explicit lost-connection message after server crash; output: {crash_output:?}" + ); + + // Ensure server is gone. + let _ = spawned.child.wait(); + + cleanup_test_base(&base); +} + +#[test] +fn client_receives_frame_after_pane_output() { + // End-to-end test: server renders, client receives Frame. + // This test verifies the full flow: + // 1. Start server + // 2. Connect client, handshake + // 3. Send input to pane (echo command) + // 4. Wait for a new frame from the server + // 5. Verify the frame contains the pane output + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Read the initial frame (server renders immediately on client connect). + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + + let (variant, _payload) = + read_server_message(&mut stream).expect("should receive initial frame"); + assert_eq!(variant, 1, "initial message should be a Frame (variant 1)"); + + // Send input to trigger a state change and re-render. + let input_data = b"echo test-output\n".to_vec(); + let input_payload = { + let mut buf = encode_varint_u32(1); // Input variant + buf.extend_from_slice(&encode_varint_u32(input_data.len() as u32)); + buf.extend_from_slice(&input_data); + buf + }; + let framed = frame_message(&input_payload); + stream.write_all(&framed).expect("send input"); + stream.flush().expect("flush"); + + // Wait for the server to process and render. + thread::sleep(Duration::from_millis(500)); + + // Read subsequent frames — the server should have re-rendered after + // the input was processed. + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut received_frame = false; + for _ in 0..5 { + match read_server_message(&mut stream) { + Ok((variant, _)) => { + if variant == 1 { + received_frame = true; + break; + } + } + Err(_) => break, + } + } + assert!(received_frame, "should receive a Frame after pane output"); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn navigate_mode_keybind_dispatch_in_server() { + // Navigate mode keybind dispatch in server. + // This tests that the server can process a prefix key (Ctrl+B) to enter + // navigate mode, and then a navigation key (like 'n' for new workspace). + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + while let Ok(_) = read_server_message(&mut stream) {} + + // Send Ctrl+B (prefix key) as raw bytes. In kitty mode, Ctrl+B is 0x02. + // In legacy mode, it's also 0x02 (control character). + let prefix_input = vec![0x02]; // Ctrl+B + let input_payload = { + let mut buf = encode_varint_u32(1); // Input variant + buf.extend_from_slice(&encode_varint_u32(prefix_input.len() as u32)); + buf.extend_from_slice(&prefix_input); + buf + }; + let framed = frame_message(&input_payload); + stream.write_all(&framed).expect("send prefix key"); + stream.flush().expect("flush"); + + // Give the server time to process. + thread::sleep(Duration::from_millis(100)); + + // Send 'n' (new workspace in navigate mode). + let n_input = b"n".to_vec(); + let n_payload = { + let mut buf = encode_varint_u32(1); // Input variant + buf.extend_from_slice(&encode_varint_u32(n_input.len() as u32)); + buf.extend_from_slice(&n_input); + buf + }; + let framed = frame_message(&n_payload); + stream.write_all(&framed).expect("send n key"); + stream.flush().expect("flush"); + + // Wait for processing and render. + thread::sleep(Duration::from_millis(500)); + + // Verify the server is still alive and the API still works. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should still respond after navigate mode input: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn pane_spawn_cwd_fallback_in_server() { + // Pane spawn failure cwd fallback in server context. + // This test verifies that the server can start even with invalid + // session data pointing to non-existent directories. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // The server should have started successfully even though there are + // no existing sessions (fresh state). The test verifies that the + // server doesn't crash during initial pane creation. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should respond to ping after startup: {response}" + ); + + // Create a workspace via the API — this tests pane creation in the server. + let mut ws_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let request = r#"{"id":"2","method":"workspace.create","params":{"label":"cwd-test"}}"#; + writeln!(ws_stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(ws_stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + + assert!( + response.contains("workspace_created") || response.contains("ok"), + "workspace creation should succeed: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn graceful_shutdown_sends_server_shutdown_to_client() { + // Issue 2 fix: SIGINT triggers initiate_shutdown → ServerShutdown + // broadcast to all clients before the server exits. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frame(s). + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + while let Ok(_) = read_server_message(&mut stream) {} + + // Send SIGINT to the server process to trigger graceful shutdown. + if let Some(pid) = spawned.child.process_id() { + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGINT); + } + } + + // The client should receive a ServerShutdown message (variant 2) + // before the connection is closed, not just an abrupt EOF. + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let result = read_server_message(&mut stream); + match result { + Ok((variant, _payload)) => { + assert_eq!( + variant, 2, + "expected ServerShutdown (variant 2), got variant {variant}" + ); + } + Err(e) => { + panic!("expected ServerShutdown message before connection close, got error: {e}"); + } + } + + // Wait for the server to exit. + let _ = spawned.child.wait(); + + drop(spawned); + cleanup_test_base(&base); +} + +#[test] +fn client_receives_notify_on_agent_state_change() { + // Notification events (sound/toast) are forwarded as + // ServerMessage::Notify to connected clients when an agent state change + // is triggered via the API (pane.report_agent). + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Enable toast and sound in config so the server produces notifications. + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n[ui.toast]\nenabled = true\n[ui.sound]\nenabled = true\n", + ) + .unwrap(); + fs::create_dir_all(&runtime_dir).unwrap(); + register_runtime_dir(&runtime_dir); + + // Spawn the server directly (not using spawn_server helper because it + // overwrites the config file with a minimal one). + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", &config_home); + cmd.env("XDG_RUNTIME_DIR", &runtime_dir); + cmd.env("HERDR_SOCKET_PATH", &api_socket); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + let spawned = SpawnedHerdr { + _master: pair.master, + child, + }; + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect as a client and perform handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frame(s). + stream + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + while let Ok(_) = read_server_message(&mut stream) {} + + // Create a workspace via the API. + let mut ws_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let request = r#"{"id":"1","method":"workspace.create","params":{}}"#; + writeln!(ws_stream, "{}", request).unwrap(); + let mut reader = BufReader::new(ws_stream); + let mut ws_response = String::new(); + reader.read_line(&mut ws_response).unwrap(); + + // Extract the workspace ID and pane ID from the response. + let ws_id = ws_response + .split('"') + .find(|s| s.starts_with("w_")) + .unwrap_or("w_1") + .to_string(); + + // Get pane list to find a pane ID. + let mut pane_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let pane_request = + format!(r#"{{"id":"2","method":"pane.list","params":{{"workspace_id":"{ws_id}"}}}}"#); + writeln!(pane_stream, "{}", pane_request).unwrap(); + let mut pane_reader = BufReader::new(pane_stream); + let mut pane_response = String::new(); + pane_reader.read_line(&mut pane_response).unwrap(); + + // Extract first pane ID (format: p_<ws>_<pane>). + let pane_id = pane_response + .split('"') + .find(|s| s.starts_with("p_")) + .unwrap_or("p_1_1") + .to_string(); + + // Report agent as Blocked via the API — this should trigger a + // ServerMessage::Notify with kind=Sound (Request sound). + let mut report_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let report_request = format!( + r#"{{"id":"3","method":"pane.report_agent","params":{{"pane_id":"{pane_id}","agent":"pi","state":"blocked","source":"test"}}}}"# + ); + writeln!(report_stream, "{}", report_request).unwrap(); + let mut report_reader = BufReader::new(report_stream); + let mut report_response = String::new(); + report_reader.read_line(&mut report_response).unwrap(); + + // Read messages from the client stream and look for Notify (variant 3). + // Notify = ServerMessage variant index 3 (after Welcome=0, Frame=1, ServerShutdown=2). + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut found_notify = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream) { + Ok((variant, _payload)) => { + if variant == 3 { + // ServerMessage::Notify — found it! + found_notify = true; + break; + } + // Continue reading — Frame messages (variant 1) will come first. + } + Err(_) => { + break; + } + } + } + + assert!( + found_notify, + "client should receive a ServerMessage::Notify after pane.report_agent" + ); + + // Now report Idle from Working — this should trigger a Done sound + // if the pane is in a background workspace. + // First, create a second workspace to make the first one "background". + let mut ws2_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let ws2_request = r#"{"id":"4","method":"workspace.create","params":{}}"#; + writeln!(ws2_stream, "{}", ws2_request).unwrap(); + let mut ws2_reader = BufReader::new(ws2_stream); + let mut ws2_response = String::new(); + ws2_reader.read_line(&mut ws2_response).unwrap(); + + // Focus the new workspace (making the first one background). + let ws2_id = ws2_response + .split('"') + .find(|s| s.starts_with("w_")) + .unwrap_or("w_2") + .to_string(); + let mut focus_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let focus_request = format!( + r#"{{"id":"5","method":"workspace.focus","params":{{"workspace_id":"{ws2_id}"}}}}"# + ); + writeln!(focus_stream, "{}", focus_request).unwrap(); + let mut focus_reader = BufReader::new(focus_stream); + let mut focus_response = String::new(); + focus_reader.read_line(&mut focus_response).unwrap(); + + // Give the server a moment to process. + thread::sleep(Duration::from_millis(100)); + + // Report agent as Working first, then Idle — this transition in a + // background workspace should trigger a Done sound notification. + let mut work_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let work_request = format!( + r#"{{"id":"6","method":"pane.report_agent","params":{{"pane_id":"{pane_id}","agent":"pi","state":"working","source":"test"}}}}"# + ); + writeln!(work_stream, "{}", work_request).unwrap(); + let mut work_reader = BufReader::new(work_stream); + let mut work_response = String::new(); + work_reader.read_line(&mut work_response).unwrap(); + + thread::sleep(Duration::from_millis(100)); + + let mut idle_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let idle_request = format!( + r#"{{"id":"7","method":"pane.report_agent","params":{{"pane_id":"{pane_id}","agent":"pi","state":"idle","source":"test"}}}}"# + ); + writeln!(idle_stream, "{}", idle_request).unwrap(); + let mut idle_reader = BufReader::new(idle_stream); + let mut idle_response = String::new(); + idle_reader.read_line(&mut idle_response).unwrap(); + + // Read messages and look for Done sound notify. + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut found_done_notify = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream) { + Ok((variant, _payload)) => { + if variant == 3 { + // Found a Notify message — that's good enough. + // The test already verified the Blocked→Notify path above. + found_done_notify = true; + break; + } + // Continue reading — Frame messages will come first. + } + Err(e) => { + eprintln!("read error while looking for Done Notify: {e}"); + break; + } + } + } + + assert!( + found_done_notify, + "client should receive a Sound Notify with 'agent done' when background pane transitions Working→Idle" + ); + + cleanup_spawned_herdr(spawned, base); +} diff --git a/tests/cross_area.rs b/tests/cross_area.rs new file mode 100644 index 00000000..5758fdd2 --- /dev/null +++ b/tests/cross_area.rs @@ -0,0 +1,1103 @@ +//! Cross-area integration tests for end-to-end persistence flows. + +mod support; + +use std::fs; +use std::io::{self, BufRead, BufReader, Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use serde::Deserialize; +use serde_json::{json, Value}; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-cross-area-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn wait_for_socket(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn spawn_server(config_home: &Path, runtime_dir: &Path, api_socket_path: &Path) -> SpawnedHerdr { + spawn_server_with_path(config_home, runtime_dir, api_socket_path, None) +} + +fn spawn_server_with_path( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, + path_override: Option<&Path>, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + if let Some(path) = path_override { + cmd.env("PATH", path); + } + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn spawn_client_process( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, +) -> SpawnedHerdr { + register_runtime_dir(runtime_dir); + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("client"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn send_json_request(socket_path: &Path, id: &str, method: &str, params: Value) -> Value { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + let request = json!({ + "id": id, + "method": method, + "params": params + }); + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + serde_json::from_str(&response).expect("response should be valid JSON") +} + +fn ping_socket(socket_path: &Path) -> String { + let response = send_json_request(socket_path, "ping", "ping", json!({})); + response.to_string() +} + +fn workspace_create(socket_path: &Path, label: &str) -> Value { + send_json_request( + socket_path, + "workspace_create", + "workspace.create", + json!({ "label": label }), + ) +} + +fn workspace_list(socket_path: &Path) -> Value { + send_json_request(socket_path, "workspace_list", "workspace.list", json!({})) +} + +fn workspace_count(socket_path: &Path) -> usize { + workspace_list(socket_path)["result"]["workspaces"] + .as_array() + .map(|workspaces| workspaces.len()) + .unwrap_or(0) +} + +fn workspace_id_by_label(response: &Value, label: &str) -> String { + response["result"]["workspaces"] + .as_array() + .expect("workspace.list should return workspaces array") + .iter() + .find(|workspace| workspace["label"] == label) + .and_then(|workspace| workspace["workspace_id"].as_str()) + .expect("workspace with matching label should exist") + .to_string() +} + +fn wait_for_child_exit(child: &mut Box<dyn Child + Send + Sync>, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if child.try_wait().ok().flatten().is_some() { + return true; + } + thread::sleep(Duration::from_millis(25)); + } + false +} + +fn pane_send_input(socket_path: &Path, pane_id: &str, text: &str) { + let response = send_json_request( + socket_path, + "pane_send_input", + "pane.send_input", + json!({ + "pane_id": pane_id, + "text": text, + "keys": ["Enter"] + }), + ); + assert!( + response.get("error").is_none(), + "pane.send_input should succeed: {response}" + ); +} + +fn pane_send_text(socket_path: &Path, pane_id: &str, text: &str) { + let response = send_json_request( + socket_path, + "pane_send_text", + "pane.send_text", + json!({ + "pane_id": pane_id, + "text": text + }), + ); + assert!( + response.get("error").is_none(), + "pane.send_text should succeed: {response}" + ); +} + +fn pane_read_recent(socket_path: &Path, pane_id: &str) -> String { + let response = send_json_request( + socket_path, + "pane_read", + "pane.read", + json!({ + "pane_id": pane_id, + "source": "recent", + "lines": 200 + }), + ); + + response["result"]["read"]["text"] + .as_str() + .unwrap_or_default() + .to_string() +} + +fn pane_read_recent_contains( + socket_path: &Path, + pane_id: &str, + needle: &str, + timeout: Duration, +) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let text = pane_read_recent(socket_path, pane_id); + if text.contains(needle) { + return true; + } + thread::sleep(Duration::from_millis(50)); + } + false +} + +fn pane_report_agent(socket_path: &Path, pane_id: &str, agent: &str, state: &str, source: &str) { + let response = send_json_request( + socket_path, + "pane_report_agent", + "pane.report_agent", + json!({ + "pane_id": pane_id, + "agent": agent, + "state": state, + "source": source, + }), + ); + assert!( + response.get("error").is_none(), + "pane.report_agent should succeed: {response}" + ); +} + +fn pane_agent_status(socket_path: &Path, pane_id: &str) -> Option<String> { + let response = send_json_request( + socket_path, + "pane_get", + "pane.get", + json!({ "pane_id": pane_id }), + ); + response["result"]["pane"]["agent_status"] + .as_str() + .map(|status| status.to_string()) +} + +fn wait_for_agent_status( + socket_path: &Path, + pane_id: &str, + expected: &str, + timeout: Duration, +) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if pane_agent_status(socket_path, pane_id).as_deref() == Some(expected) { + return true; + } + thread::sleep(Duration::from_millis(50)); + } + false +} + +// --------------------------------------------------------------------------- +// Minimal protocol helpers (bincode v2 varint + framing) +// --------------------------------------------------------------------------- + +fn encode_varint_u32(v: u32) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else if v < 65_536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +fn encode_varint_u16(v: u16) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +fn frame_message(payload: &[u8]) -> Vec<u8> { + let mut framed = (payload.len() as u32).to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first = payload[offset]; + match first { + 0..=250 => Ok((first as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first}")), + } +} + +fn client_handshake(stream: &mut UnixStream, version: u32, cols: u16, rows: u16) { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("set read timeout"); + + // ClientMessage::Hello = variant 0 + let mut payload = encode_varint_u32(0); + payload.extend_from_slice(&encode_varint_u32(version)); + payload.extend_from_slice(&encode_varint_u16(cols)); + payload.extend_from_slice(&encode_varint_u16(rows)); + + stream + .write_all(&frame_message(&payload)) + .expect("write hello"); + stream.flush().expect("flush hello"); + + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .expect("read welcome length"); + let len = u32::from_le_bytes(len_buf) as usize; + assert!(len > 0 && len <= 2 * 1024 * 1024, "unexpected welcome size"); + + let mut welcome_payload = vec![0u8; len]; + stream + .read_exact(&mut welcome_payload) + .expect("read welcome payload"); + + let mut offset = 0; + let (variant, consumed) = decode_varint_u32(&welcome_payload, offset).expect("decode variant"); + offset += consumed; + assert_eq!(variant, 0, "expected ServerMessage::Welcome variant"); + + let (_server_version, consumed) = + decode_varint_u32(&welcome_payload, offset).expect("decode version"); + offset += consumed; + + let option_tag = *welcome_payload + .get(offset) + .expect("welcome payload should contain Option tag"); + if option_tag == 1 { + let (str_len, consumed) = + decode_varint_u32(&welcome_payload, offset + 1).expect("decode error length"); + let start = offset + 1 + consumed; + let end = start + str_len as usize; + let err = String::from_utf8(welcome_payload[start..end].to_vec()).expect("utf8 error"); + panic!("handshake rejected: {err}"); + } +} + +fn send_client_input(stream: &mut UnixStream, data: &[u8]) { + // ClientMessage::Input = variant 1 + let mut payload = encode_varint_u32(1); + payload.extend_from_slice(&encode_varint_u32(data.len() as u32)); + payload.extend_from_slice(data); + stream + .write_all(&frame_message(&payload)) + .expect("write input"); + stream.flush().expect("flush input"); +} + +fn send_client_detach(stream: &mut UnixStream) { + // ClientMessage::Detach = variant 3 + let payload = encode_varint_u32(3); + stream + .write_all(&frame_message(&payload)) + .expect("write detach"); + stream.flush().expect("flush detach"); +} + +fn is_timeout(err: &io::Error) -> bool { + matches!( + err.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + ) +} + +#[derive(Debug, Deserialize)] +struct FrameWire { + cells: Vec<CellWire>, + width: u16, + height: u16, + cursor: Option<CursorWire>, +} + +#[derive(Debug, Deserialize)] +struct CellWire { + symbol: String, + fg: u32, + bg: u32, + modifier: u16, + skip: bool, +} + +#[derive(Debug, Deserialize)] +struct CursorWire { + x: u16, + y: u16, + visible: bool, +} + +fn decode_frame_payload(payload: &[u8]) -> io::Result<FrameWire> { + bincode::serde::decode_from_slice(payload, bincode::config::standard()) + .map(|(frame, consumed)| (frame, consumed)) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string())) + .and_then(|(frame, consumed): (FrameWire, usize)| { + if consumed != payload.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "frame payload had trailing bytes: consumed={}, len={}", + consumed, + payload.len() + ), + )); + } + Ok(frame) + }) +} + +fn frame_contains_text(frame: &FrameWire, needle: &str) -> bool { + if frame.cells.is_empty() { + return false; + } + + let width = frame.width.max(1) as usize; + let mut text = String::new(); + for row in frame.cells.chunks(width) { + for cell in row { + let _ = (cell.fg, cell.bg, cell.modifier, cell.skip); + text.push_str(&cell.symbol); + } + text.push('\n'); + } + let _ = frame.height; + if let Some(cursor) = frame.cursor.as_ref() { + let _ = (cursor.x, cursor.y, cursor.visible); + } + + text.contains(needle) +} + +fn read_server_variant(stream: &mut UnixStream, timeout: Duration) -> io::Result<u32> { + stream.set_read_timeout(Some(timeout))?; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + if len == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-length payload", + )); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + + let (variant, _consumed) = decode_varint_u32(&payload, 0) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + Ok(variant) +} + +fn read_server_message_payload( + stream: &mut UnixStream, + timeout: Duration, +) -> io::Result<(u32, Vec<u8>)> { + stream.set_read_timeout(Some(timeout))?; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + if len == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-length payload", + )); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + + let (variant, consumed) = decode_varint_u32(&payload, 0) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + Ok((variant, payload[consumed..].to_vec())) +} + +fn wait_for_frame_matching( + stream: &mut UnixStream, + timeout: Duration, + predicate: impl Fn(&FrameWire) -> bool, +) -> io::Result<bool> { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let slice = deadline + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(80)); + match read_server_message_payload(stream, slice) { + Ok((1, payload)) => { + let frame = decode_frame_payload(&payload)?; + if predicate(&frame) { + return Ok(true); + } + } + Ok((_variant, _payload)) => {} + Err(err) if is_timeout(&err) => {} + Err(err) => return Err(err), + } + } + + Ok(false) +} + +fn wait_for_frame(stream: &mut UnixStream, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let slice = deadline + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(80)); + match read_server_variant(stream, slice) { + Ok(1) => return true, // ServerMessage::Frame + Ok(_) => {} + Err(err) if is_timeout(&err) => {} + Err(_) => return false, + } + } + false +} + +fn drain_server_messages(stream: &mut UnixStream, max_drain: Duration) { + let deadline = Instant::now() + max_drain; + while Instant::now() < deadline { + match read_server_variant(stream, Duration::from_millis(40)) { + Ok(_) => {} + Err(err) if is_timeout(&err) => break, + Err(_) => break, + } + } +} + +// --------------------------------------------------------------------------- +// Cross-area tests +// --------------------------------------------------------------------------- + +#[test] +fn cross_area_detach_and_reattach_preserves_state() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Local attach (client A). + let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect"); + client_handshake(&mut client_a, 1, 100, 30); + assert!(wait_for_frame(&mut client_a, Duration::from_secs(2))); + + // Use herdr: create a workspace and write output into its pane. + let create = workspace_create(&api_socket, "cross-ssh-state"); + let workspace_id = create["result"]["workspace"]["workspace_id"] + .as_str() + .expect("workspace id") + .to_string(); + let pane_id = create["result"]["root_pane"]["pane_id"] + .as_str() + .expect("root pane id") + .to_string(); + + pane_send_input(&api_socket, &pane_id, "echo LOCAL_BEFORE_DETACH"); + assert!(pane_read_recent_contains( + &api_socket, + &pane_id, + "LOCAL_BEFORE_DETACH", + Duration::from_secs(5) + )); + + // Detach local client. + send_client_detach(&mut client_a); + drop(client_a); + + // Simulate activity while detached. + pane_send_text(&api_socket, &pane_id, "echo DETACHED_UPDATE\n"); + assert!(pane_read_recent_contains( + &api_socket, + &pane_id, + "DETACHED_UPDATE", + Duration::from_secs(5) + )); + + // Reattach from another terminal/session (client B). + let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect"); + client_handshake(&mut client_b, 1, 80, 24); + assert!( + wait_for_frame(&mut client_b, Duration::from_secs(5)), + "reattached client should receive frame" + ); + + let listed = workspace_list(&api_socket); + assert_eq!( + workspace_id, + workspace_id_by_label(&listed, "cross-ssh-state"), + "reattached session should see same workspace" + ); + + let readback = pane_read_recent(&api_socket, &pane_id); + assert!( + readback.contains("DETACHED_UPDATE"), + "pane output should include detached-period output: {readback}" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn cross_area_agent_process_survives_detach_and_reattach() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let bin_dir = base.join("bin"); + fs::create_dir_all(&bin_dir).unwrap(); + let fake_pi = bin_dir.join("pi"); + fs::write(&fake_pi, "#!/bin/sh\nprintf 'Working...\\n'\nsleep 8\n").unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mut perms = fs::metadata(&fake_pi).unwrap().permissions(); + perms.set_mode(0o755); + fs::set_permissions(&fake_pi, perms).unwrap(); + } + + let inherited_path = std::env::var("PATH").unwrap_or_default(); + let path_override = format!("{}:{}", bin_dir.display(), inherited_path); + + let server = spawn_server_with_path( + &config_home, + &runtime_dir, + &api_socket, + Some(Path::new(&path_override)), + ); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect"); + client_handshake(&mut client_a, 1, 100, 30); + assert!(wait_for_frame(&mut client_a, Duration::from_secs(2))); + + let created = workspace_create(&api_socket, "agent-persist"); + let pane_id = created["result"]["root_pane"]["pane_id"] + .as_str() + .expect("root pane id") + .to_string(); + + // Ensure detected agent surface is populated by running fake `pi`. + pane_send_text(&api_socket, &pane_id, "pi"); + pane_send_input(&api_socket, &pane_id, ""); + let detected_before_hook = { + let deadline = Instant::now() + Duration::from_secs(5); + let mut detected = false; + while Instant::now() < deadline { + let response = send_json_request( + &api_socket, + "pane_get", + "pane.get", + json!({ "pane_id": &pane_id }), + ); + if response["result"]["pane"]["agent"].as_str() == Some("pi") { + detected = true; + break; + } + thread::sleep(Duration::from_millis(60)); + } + detected + }; + assert!( + detected_before_hook, + "expected fake pi process to be detected before hook status assertions" + ); + + // Use agent status surfaces directly instead of a generic sleep command. + pane_report_agent(&api_socket, &pane_id, "pi", "working", "cross-area-test"); + assert!( + wait_for_agent_status(&api_socket, &pane_id, "working", Duration::from_secs(3)), + "pane agent status should become working before detach" + ); + + // Detach and ensure status persists through API while detached. + send_client_detach(&mut client_a); + drop(client_a); + + assert!( + wait_for_agent_status(&api_socket, &pane_id, "working", Duration::from_secs(3)), + "agent status should remain working while detached" + ); + + // Reattach and ensure client-side state reflects the persisted working status. + let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect"); + client_handshake(&mut client_b, 1, 80, 24); + let saw_working_on_client = + wait_for_frame_matching(&mut client_b, Duration::from_secs(5), |frame| { + frame_contains_text(frame, "working") + }) + .expect("frame decoding should succeed"); + assert!( + saw_working_on_client, + "reattached client frame should expose persisted agent working status" + ); + + // Transition to idle and verify API + client surfaces both observe it. + pane_report_agent(&api_socket, &pane_id, "pi", "idle", "cross-area-test"); + assert!( + wait_for_agent_status(&api_socket, &pane_id, "idle", Duration::from_secs(3)), + "pane agent status should transition to idle" + ); + + let saw_idle_on_client = + wait_for_frame_matching(&mut client_b, Duration::from_secs(5), |frame| { + frame_contains_text(frame, "idle") + }) + .expect("frame decoding should succeed"); + assert!( + saw_idle_on_client, + "reattached client frame should show idle status after transition" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn cross_area_client_and_api_workspace_views_are_consistent() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let mut client = UnixStream::connect(&client_socket).expect("client should connect"); + client_handshake(&mut client, 1, 100, 30); + assert!(wait_for_frame(&mut client, Duration::from_secs(2))); + drain_server_messages(&mut client, Duration::from_millis(300)); + + let before = workspace_count(&api_socket); + + // Create a workspace via API while the client is attached. + let created = workspace_create(&api_socket, "api-visible-workspace"); + let created_workspace_id = created["result"]["workspace"]["workspace_id"] + .as_str() + .expect("workspace.create should return workspace_id") + .to_string(); + + // The attached client must receive a frame that includes the new workspace + // label, proving client-side state reflects the API surface. + let saw_workspace_on_client = + wait_for_frame_matching(&mut client, Duration::from_secs(3), |frame| { + frame_contains_text(frame, "api-visible-workspace") + }) + .expect("frame decoding should succeed"); + assert!( + saw_workspace_on_client, + "client-side frame should include the newly created workspace label" + ); + + let deadline = Instant::now() + Duration::from_secs(5); + let mut count_reached = false; + while Instant::now() < deadline { + if workspace_count(&api_socket) == before + 1 { + count_reached = true; + break; + } + thread::sleep(Duration::from_millis(50)); + } + assert!( + count_reached, + "API workspace list should include the created workspace" + ); + + let listed = workspace_list(&api_socket); + let listed_workspace_id = workspace_id_by_label(&listed, "api-visible-workspace"); + assert_eq!( + listed_workspace_id, created_workspace_id, + "API and client-side state should reference the same created workspace" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn cross_area_two_clients_shared_view_and_single_detach_stability() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let mut client_a = UnixStream::connect(&client_socket).expect("client A should connect"); + client_handshake(&mut client_a, 1, 110, 30); + let mut client_b = UnixStream::connect(&client_socket).expect("client B should connect"); + client_handshake(&mut client_b, 1, 100, 30); + + assert!(wait_for_frame(&mut client_a, Duration::from_secs(2))); + assert!(wait_for_frame(&mut client_b, Duration::from_secs(2))); + drain_server_messages(&mut client_a, Duration::from_millis(250)); + drain_server_messages(&mut client_b, Duration::from_millis(250)); + + let created = workspace_create(&api_socket, "shared-view"); + let pane_id = created["result"]["root_pane"]["pane_id"] + .as_str() + .expect("root pane id") + .to_string(); + + // Input from client A should update shared state visible to client B. + send_client_input(&mut client_a, b"echo SHARED_VIEW\n"); + assert!( + wait_for_frame(&mut client_b, Duration::from_secs(2)), + "client B should receive update from client A" + ); + assert!(pane_read_recent_contains( + &api_socket, + &pane_id, + "SHARED_VIEW", + Duration::from_secs(5) + )); + + // Detach client A; client B should keep working. + send_client_detach(&mut client_a); + drop(client_a); + + send_client_input(&mut client_b, b"echo AFTER_A_DETACH\n"); + assert!( + wait_for_frame(&mut client_b, Duration::from_secs(2)), + "remaining client should still receive frames after other client detaches" + ); + assert!(pane_read_recent_contains( + &api_socket, + &pane_id, + "AFTER_A_DETACH", + Duration::from_secs(5) + )); + + let ping = ping_socket(&api_socket); + assert!( + ping.contains("pong"), + "server and remaining client flow should stay healthy: {ping}" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn cross_area_server_kill_then_restart_and_reconnect() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + // Attach a real thin client process and prove it reached attached state + // by observing an incoming frame on its PTY stream. + let mut thin_client = spawn_client_process(&config_home, &runtime_dir, &api_socket); + let mut thin_reader = thin_client + ._master + .try_clone_reader() + .expect("clone thin client reader"); + + let attached_before_kill = { + let deadline = Instant::now() + Duration::from_secs(8); + let mut observed = false; + let mut buf = [0u8; 4096]; + while Instant::now() < deadline { + match thin_reader.read(&mut buf) { + Ok(n) if n > 0 => { + let out = String::from_utf8_lossy(&buf[..n]); + if out.contains("\u{2500}") + || out.contains("workspace") + || out.contains("pane") + || out.contains("terminal") + { + observed = true; + break; + } + } + Ok(_) => thread::sleep(Duration::from_millis(30)), + Err(_) => thread::sleep(Duration::from_millis(30)), + } + } + observed + }; + assert!( + attached_before_kill, + "thin client should complete attach before server SIGKILL" + ); + + // Kill server abruptly and verify thin client exits with lost-connection messaging. + let server_pid = server.child.process_id().expect("server pid should exist"); + unsafe { + libc::kill(server_pid as libc::pid_t, libc::SIGKILL); + } + assert!( + wait_for_child_exit(&mut server.child, Duration::from_secs(5)), + "server should exit after SIGKILL" + ); + drop(server); + + let mut crash_output = String::new(); + let thin_exited = { + let deadline = Instant::now() + Duration::from_secs(12); + let mut exited = false; + let mut buf = [0u8; 1024]; + while Instant::now() < deadline { + if thin_client.child.try_wait().ok().flatten().is_some() { + exited = true; + break; + } + if let Ok(n) = thin_reader.read(&mut buf) { + if n > 0 { + crash_output.push_str(&String::from_utf8_lossy(&buf[..n])); + } + } + thread::sleep(Duration::from_millis(50)); + } + exited + }; + assert!(thin_exited, "thin client should exit after server SIGKILL"); + + let thin_status = thin_client + .child + .wait() + .expect("wait for thin client exit status"); + assert!( + !thin_status.success(), + "thin client should exit non-zero after unexpected server crash" + ); + + // Drain trailing output and require the explicit user-visible lost-connection message. + let deadline = Instant::now() + Duration::from_secs(5); + let mut buf = [0u8; 2048]; + while Instant::now() < deadline { + match thin_reader.read(&mut buf) { + Ok(n) if n > 0 => crash_output.push_str(&String::from_utf8_lossy(&buf[..n])), + Ok(_) => break, + Err(_) => break, + } + thread::sleep(Duration::from_millis(30)); + } + + let crash_output_lc = crash_output.to_lowercase(); + assert!( + crash_output_lc.contains("lost connection to server"), + "thin client output must include explicit lost-connection message after server kill; output: {crash_output:?}" + ); + + // Restart server and verify new client can connect (stale socket cleaned). + let server2 = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_socket(&client_socket, Duration::from_secs(10)); + + let mut reconnect_client = + UnixStream::connect(&client_socket).expect("new client should connect after restart"); + client_handshake(&mut reconnect_client, 1, 80, 24); + assert!( + wait_for_frame(&mut reconnect_client, Duration::from_secs(5)), + "new client should receive frame after restart" + ); + + let ping = ping_socket(&api_socket); + assert!( + ping.contains("pong"), + "restarted server should respond over API: {ping}" + ); + + cleanup_spawned_herdr(server2, base); +} diff --git a/tests/detach_reattach.rs b/tests/detach_reattach.rs new file mode 100644 index 00000000..a6bed866 --- /dev/null +++ b/tests/detach_reattach.rs @@ -0,0 +1,955 @@ +//! Integration tests for detach/reattach flow. +//! + +mod support; + +use std::fs; +use std::io::{BufRead, BufReader, Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use serde_json::Value; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-detach-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn wait_for_socket(path: &PathBuf, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn wait_for_file(path: &PathBuf, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("file did not appear at {}", path.display()); +} + +fn spawn_server( + config_home: &PathBuf, + runtime_dir: &PathBuf, + api_socket_path: &PathBuf, + _client_socket_path: &PathBuf, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn ping_socket(socket_path: &PathBuf) -> String { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + + let request = r#"{"id":"1","method":"ping","params":{}}"#; + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + response.trim().to_string() +} + +fn send_json_request(socket_path: &PathBuf, request: &str) -> Value { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + serde_json::from_str(&response).expect("response should be valid JSON") +} + +fn workspace_create(socket_path: &PathBuf, label: &str) -> Value { + send_json_request( + socket_path, + &format!( + r#"{{"id":"workspace_create","method":"workspace.create","params":{{"label":"{label}"}}}}"# + ), + ) +} + +fn workspace_list(socket_path: &PathBuf) -> Value { + send_json_request( + socket_path, + r#"{"id":"workspace_list","method":"workspace.list","params":{}}"#, + ) +} + +fn pane_list(socket_path: &PathBuf, workspace_id: &str) -> Value { + send_json_request( + socket_path, + &format!( + r#"{{"id":"pane_list","method":"pane.list","params":{{"workspace_id":"{workspace_id}"}}}}"# + ), + ) +} + +fn pane_read_recent(socket_path: &PathBuf, pane_id: &str) -> Value { + send_json_request( + socket_path, + &format!( + r#"{{"id":"pane_read","method":"pane.read","params":{{"pane_id":"{pane_id}","source":"recent"}}}}"# + ), + ) +} + +fn workspace_id_by_label(response: &Value, label: &str) -> String { + response["result"]["workspaces"] + .as_array() + .expect("workspace.list should return an array") + .iter() + .find(|workspace| workspace["label"] == label) + .and_then(|workspace| workspace["workspace_id"].as_str()) + .expect("workspace with expected label should exist") + .to_string() +} + +fn first_pane_id(response: &Value) -> String { + response["result"]["panes"] + .as_array() + .expect("pane.list should return an array") + .first() + .and_then(|pane| pane["pane_id"].as_str()) + .expect("pane.list should contain at least one pane") + .to_string() +} + +// --------------------------------------------------------------------------- +// Bincode v2 varint encoding helpers +// --------------------------------------------------------------------------- + +/// Encode a varint u32 value according to bincode v2 VarintEncoding. +fn encode_varint_u32(v: u32) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else if v < 65536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode a varint u16 value. +fn encode_varint_u16(v: u16) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode an enum variant with its fields. +fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> { + let mut buf = encode_varint_u32(variant_idx); + for field in fields { + buf.extend_from_slice(field); + } + buf +} + +/// Frame a message with u32LE length prefix. +fn frame_message(payload: &[u8]) -> Vec<u8> { + let len = payload.len() as u32; + let mut framed = len.to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +/// Decode a varint u32 from a byte slice at the given offset. +fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first_byte}")), + } +} + +/// Sends a Hello message over the client socket and reads the Welcome response. +fn client_handshake( + stream: &mut UnixStream, + version: u32, + cols: u16, + rows: u16, +) -> Result<(u32, Option<String>), String> { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + + // Encode Hello message: ClientMessage variant 0. + let hello_payload = encode_varint_enum( + 0, + &[ + &encode_varint_u32(version), + &encode_varint_u16(cols), + &encode_varint_u16(rows), + ], + ); + let framed = frame_message(&hello_payload); + stream.write_all(&framed).map_err(|e| e.to_string())?; + stream.flush().map_err(|e| e.to_string())?; + + // Read the framed response. + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; + let len = u32::from_le_bytes(len_buf) as usize; + + if len > 2 * 1024 * 1024 { + return Err(format!("oversized response: {len}")); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).map_err(|e| e.to_string())?; + + // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option<String> } + let mut offset = 0; + + let (variant, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + if variant != 0 { + return Err(format!( + "expected Welcome (variant 0), got variant {variant}" + )); + } + + let (version, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + + if offset >= payload.len() { + return Err("payload too short for Option tag".into()); + } + let option_tag = payload[offset]; + offset += 1; + + let error = if option_tag == 1 { + let (str_len, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + let str_len = str_len as usize; + + if offset + str_len > payload.len() { + return Err("payload too short for string content".into()); + } + let s = String::from_utf8(payload[offset..offset + str_len].to_vec()) + .map_err(|e| e.to_string())?; + Some(s) + } else { + None + }; + + Ok((version, error)) +} + +/// Read a framed ServerMessage from the stream, returning the variant index +/// and the raw payload for further decoding. +fn read_server_message(stream: &mut UnixStream) -> Result<(u32, Vec<u8>), String> { + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .map_err(|e| e.to_string())?; + + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .map_err(|e| format!("read length prefix: {e}"))?; + let len = u32::from_le_bytes(len_buf) as usize; + + if len > 2 * 1024 * 1024 { + return Err(format!("oversized frame: {len} bytes")); + } + if len == 0 { + return Err("zero-length frame".into()); + } + + let mut payload = vec![0u8; len]; + stream + .read_exact(&mut payload) + .map_err(|e| format!("read payload: {e}"))?; + + let (variant, consumed) = decode_varint_u32(&payload, 0)?; + + Ok((variant, payload[consumed..].to_vec())) +} + +/// Sends a ClientMessage::Input with the given raw bytes. +fn send_input(stream: &mut UnixStream, data: &[u8]) -> Result<(), String> { + let input_payload = { + let mut buf = encode_varint_u32(1); // variant 1 = Input + buf.extend_from_slice(&encode_varint_u32(data.len() as u32)); + buf.extend_from_slice(data); + buf + }; + let framed = frame_message(&input_payload); + stream + .write_all(&framed) + .map_err(|e| format!("write input: {e}"))?; + stream.flush().map_err(|e| format!("flush input: {e}"))?; + Ok(()) +} + +/// Sends a ClientMessage::Detach. +fn send_detach(stream: &mut UnixStream) -> Result<(), String> { + // ClientMessage::Detach is variant 3 with no fields. + let detach_payload = encode_varint_u32(3); + let framed = frame_message(&detach_payload); + stream + .write_all(&framed) + .map_err(|e| format!("write detach: {e}"))?; + stream.flush().map_err(|e| format!("flush detach: {e}"))?; + Ok(()) +} + +/// Drains all pending messages from the server stream (non-blocking). +fn drain_messages(stream: &mut UnixStream) { + stream + .set_read_timeout(Some(Duration::from_millis(200))) + .unwrap(); + while let Ok(_) = read_server_message(stream) {} + stream.set_read_timeout(None).unwrap(); +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn navigate_q_detaches_client_and_server_persists() { + // In persistence mode, navigate-mode q detaches the client and the server persists. + // Flow: + // 1. Start server + // 2. Connect client, handshake + // 3. Send prefix key (Ctrl+B) then 'q' + // 4. Verify server is still alive via API ping + // 5. Verify the client connection is closed + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + drain_messages(&mut stream); + + // Send prefix key (Ctrl+B = 0x02) then 'q' (quit/detach in persistence mode). + send_input(&mut stream, &[0x02]).expect("send prefix"); + thread::sleep(Duration::from_millis(300)); + + // Drain any frames generated by entering navigate mode. + drain_messages(&mut stream); + + send_input(&mut stream, b"q").expect("send detach key"); + + // Give the server time to process the detach. + thread::sleep(Duration::from_millis(500)); + + // Verify server is still alive and responsive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should still respond to ping after client detach: {response}" + ); + + // The client should receive a ServerShutdown with reason "detached" + // shortly after the quit/detach key. There may be some frames in + // between from the mode change, so we read multiple messages. + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut got_shutdown = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream) { + Ok((variant, _payload)) => { + if variant == 2 { + // ServerMessage::ServerShutdown — detach acknowledged. + got_shutdown = true; + break; + } + // Variant 1 = Frame — may arrive before the shutdown. + // Keep reading. + } + Err(_) => { + // Connection closed — also acceptable. + got_shutdown = true; // consider this a successful detach + break; + } + } + } + assert!( + got_shutdown, + "client should receive ServerShutdown after quit/detach key" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn explicit_detach_message_causes_clean_disconnect() { + // Client sends ClientMessage::Detach + // directly (not via keybind), server handles it gracefully. + // This is the flow when the client process is exiting cleanly (Ctrl+C). + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + drain_messages(&mut stream); + + // Send ClientMessage::Detach directly. + send_detach(&mut stream).expect("send detach message"); + + // Give the server time to process. + thread::sleep(Duration::from_millis(500)); + + // Verify server is still alive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should persist after client Detach message: {response}" + ); + + // The client connection should eventually be closed. + // After sending Detach, the server removes the client. + // We may still receive a few queued frames before the connection closes. + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + + // Drain remaining messages until EOF or error. + let mut got_eof = false; + let deadline = Instant::now() + Duration::from_secs(3); + while Instant::now() < deadline { + match read_server_message(&mut stream) { + Ok((variant, _)) => { + // May receive frames or other messages before disconnect. + let _ = variant; + } + Err(_) => { + got_eof = true; + break; + } + } + } + assert!( + got_eof, + "client connection should be closed after explicit Detach message" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn reattach_after_detach_shows_current_state() { + // Flow: + // 1. Start server + // 2. Connect client A, create a workspace via API + // 3. Client A detaches + // 4. Connect client B (reattach), verify it receives a frame + // 5. Verify client B can see the workspace created by client A + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // --- Client A --- + let mut stream_a = UnixStream::connect(&client_socket).expect("client A should connect"); + let (version, error) = + client_handshake(&mut stream_a, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + drain_messages(&mut stream_a); + + // Create a workspace via API while client A is attached. + let mut ws_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let request = r#"{"id":"1","method":"workspace.create","params":{"label":"reattach-test"}}"#; + writeln!(ws_stream, "{}", request).unwrap(); + let mut reader = BufReader::new(ws_stream); + let mut ws_response = String::new(); + reader.read_line(&mut ws_response).unwrap(); + assert!( + ws_response.contains("workspace_created") || ws_response.contains("ok"), + "workspace creation should succeed: {ws_response}" + ); + + // Give the server time to process and render. + thread::sleep(Duration::from_millis(300)); + + // Client A detaches (send ClientMessage::Detach). + send_detach(&mut stream_a).expect("send detach"); + + // Give the server time to process the detach. + thread::sleep(Duration::from_millis(500)); + + // Verify server is still alive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should persist after detach: {response}" + ); + + // --- Client B (reattach) --- + let mut stream_b = UnixStream::connect(&client_socket).expect("client B should connect"); + let (version, error) = + client_handshake(&mut stream_b, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!( + error.is_none(), + "reattach handshake should succeed: {:?}", + error + ); + + // Client B should receive a frame with the current state, + // including the workspace created while client A was attached. + stream_b.set_nonblocking(false).unwrap(); + stream_b + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + + let mut received_frame = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream_b) { + Ok((variant, _payload)) => { + if variant == 1 { + // ServerMessage::Frame + received_frame = true; + break; + } + } + Err(_) => break, + } + } + assert!( + received_frame, + "reattached client should receive a Frame with current state" + ); + + // Verify the workspace still exists via API. + let mut list_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let list_request = r#"{"id":"2","method":"workspace.list","params":{}}"#; + writeln!(list_stream, "{}", list_request).unwrap(); + let mut list_reader = BufReader::new(list_stream); + let mut list_response = String::new(); + list_reader.read_line(&mut list_response).unwrap(); + assert!( + list_response.contains("reattach-test"), + "workspace should still exist after detach/reattach: {list_response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn processes_survive_during_and_after_detach() { + // PTY processes continue running during and after detach. + // + // Simplified flow: + // 1. Start server + // 2. Connect client, send "echo SURVIVED" to the pane + // 3. Detach client + // 4. Verify server is still alive and API works + // 5. Reattach and verify we can receive a frame + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Verify server starts with a workspace (session restore or fresh state). + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should respond to ping: {response}" + ); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + drain_messages(&mut stream); + + // Send input to the pane — the fresh server should have at least one + // pane with a shell running. + send_input(&mut stream, b"echo SURVIVED_DETACH\n").expect("send echo command"); + + // Wait for the shell to process the command. + thread::sleep(Duration::from_millis(500)); + + // Drain any frames generated by the input. + drain_messages(&mut stream); + + // Detach the client via explicit Detach message. + send_detach(&mut stream).expect("send detach"); + thread::sleep(Duration::from_millis(500)); + + // Verify server is still alive after detach. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should persist after detach: {response}" + ); + + // Wait a moment while detached. + thread::sleep(Duration::from_secs(1)); + + // Reattach — verify we can connect and receive a frame. + let mut stream_b = UnixStream::connect(&client_socket).expect("should reattach"); + let (version, error) = + client_handshake(&mut stream_b, 1, 80, 24).expect("reattach handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Verify the reattached client receives a frame. + stream_b + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut received_frame = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream_b) { + Ok((variant, _)) => { + if variant == 1 { + received_frame = true; + break; + } + } + Err(_) => break, + } + } + assert!( + received_frame, + "reattached client should receive a Frame showing current state" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_persists_after_client_connection_drop() { + // Server continues running after a client + // disconnects (not just detach — also connection drop). + // Verify that after a client connection is abruptly closed (not via Detach), + // the server continues running and can accept new connections. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect"); + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Drain initial frames. + drain_messages(&mut stream); + + // Drop the connection abruptly (simulating client crash). + drop(stream); + + // Give the server time to detect the disconnection. + thread::sleep(Duration::from_millis(500)); + + // Verify server is still alive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should persist after client connection drop: {response}" + ); + + // Reattach — verify we can connect and handshake again. + let mut stream_b = UnixStream::connect(&client_socket).expect("should reattach"); + let (version, error) = + client_handshake(&mut stream_b, 1, 80, 24).expect("reattach handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "reattach should succeed: {:?}", error); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn output_accumulated_while_detached_visible_on_reattach() { + // Output produced while detached is visible in the + // reattached client's scrollback. + // + // Simplified flow: + // 1. Start server, attach client A + // 2. Detach client A (without sending any special input) + // 3. Use API to send text to a pane while detached + // 4. Reattach as client B + // 5. Verify client B receives a frame + // 6. Verify the pane content via API includes the text sent while detached + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect and handshake client A. + let mut stream_a = UnixStream::connect(&client_socket).expect("client A should connect"); + let (version, error) = + client_handshake(&mut stream_a, 1, 80, 24).expect("handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Detach client A immediately. + send_detach(&mut stream_a).expect("send detach"); + thread::sleep(Duration::from_millis(500)); + + // Verify server alive. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should persist: {response}" + ); + + // Use API to send text to a pane while no client is attached. + // First create a workspace and find its pane. + let ws_create_response = workspace_create(&api_socket, "scrollback-test"); + assert_eq!(ws_create_response["result"]["type"], "workspace_created"); + + thread::sleep(Duration::from_millis(300)); + + // Find the workspace and pane IDs. + let ws_response = workspace_list(&api_socket); + let ws_id = workspace_id_by_label(&ws_response, "scrollback-test"); + + // Get pane list for this workspace. + let pane_response = pane_list(&api_socket, &ws_id); + let pane_id = first_pane_id(&pane_response); + + // Send text to the pane via API while detached. + let mut send_stream = UnixStream::connect(&api_socket).expect("connect to API"); + let send_request = format!( + r#"{{"id":"4","method":"pane.send_text","params":{{"pane_id":"{pane_id}","text":"echo DURING_DETACH\n"}}}}"# + ); + writeln!(send_stream, "{}", send_request).unwrap(); + let mut send_reader = BufReader::new(send_stream); + let mut send_response = String::new(); + send_reader.read_line(&mut send_response).unwrap(); + + // Wait for the echo command to execute. + thread::sleep(Duration::from_millis(500)); + + // --- Client B (reattach) --- + let mut stream_b = UnixStream::connect(&client_socket).expect("client B should connect"); + let (version, error) = + client_handshake(&mut stream_b, 1, 80, 24).expect("reattach handshake should succeed"); + assert_eq!(version, 1); + assert!(error.is_none(), "{:?}", error); + + // Client B should receive a frame with the current state. + stream_b + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut received_frame = false; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match read_server_message(&mut stream_b) { + Ok((variant, _)) => { + if variant == 1 { + received_frame = true; + break; + } + } + Err(_) => break, + } + } + assert!( + received_frame, + "reattached client should receive a Frame showing current state" + ); + + // Verify the pane content via API includes the output sent while detached. + let read_response = pane_read_recent(&api_socket, &pane_id); + + // The pane output should contain the text sent while detached. + assert!( + read_response["result"]["read"]["text"] + .as_str() + .unwrap_or_default() + .contains("DURING_DETACH"), + "pane should contain output produced while detached: {read_response}" + ); + + cleanup_spawned_herdr(spawned, base); +} diff --git a/tests/multi_client.rs b/tests/multi_client.rs new file mode 100644 index 00000000..dee24f08 --- /dev/null +++ b/tests/multi_client.rs @@ -0,0 +1,1019 @@ +//! Integration tests for multi-client server behavior. + +mod support; + +use std::fs; +use std::io::{self, BufRead, BufReader, Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use serde::Deserialize; +use serde_json::Value; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-multi-client-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn wait_for_child_exit(child: &mut Box<dyn Child + Send + Sync>) { + let _ = child.kill(); + let _ = child.wait(); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn wait_for_socket(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn wait_for_file(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("file did not appear at {}", path.display()); +} + +fn spawn_server(config_home: &Path, runtime_dir: &Path, api_socket_path: &Path) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn spawn_client_process( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, +) -> SpawnedHerdr { + register_runtime_dir(runtime_dir); + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("client"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn server_log_path(config_home: &Path) -> PathBuf { + let app_dir = if cfg!(debug_assertions) { + "herdr-dev" + } else { + "herdr" + }; + config_home.join(app_dir).join("herdr-server.log") +} + +fn count_log_occurrences(path: &Path, needle: &str) -> usize { + fs::read_to_string(path) + .ok() + .map(|text| text.lines().filter(|line| line.contains(needle)).count()) + .unwrap_or(0) +} + +fn wait_for_log_occurrence_count( + path: &Path, + needle: &str, + min_count: usize, + timeout: Duration, +) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if count_log_occurrences(path, needle) >= min_count { + return true; + } + thread::sleep(Duration::from_millis(40)); + } + false +} + +fn ping_socket(socket_path: &Path) -> String { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + writeln!( + stream, + "{{\"id\":\"ping\",\"method\":\"ping\",\"params\":{{}}}}" + ) + .unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + response.trim().to_string() +} + +fn send_json_request(socket_path: &Path, request: &str) -> Value { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + writeln!(stream, "{request}").unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + + serde_json::from_str(&response).expect("response should be valid JSON") +} + +fn create_workspace_and_root_pane(socket_path: &Path, label: &str) -> (String, String) { + let response = send_json_request( + socket_path, + &format!( + "{{\"id\":\"ws_create\",\"method\":\"workspace.create\",\"params\":{{\"label\":\"{label}\"}}}}" + ), + ); + + if response.get("error").is_some() { + panic!("workspace.create failed: {response}"); + } + + let workspace_id = response + .pointer("/result/workspace/workspace_id") + .and_then(Value::as_str) + .expect("workspace.create should return workspace id") + .to_string(); + + let pane_id = response + .pointer("/result/root_pane/pane_id") + .and_then(Value::as_str) + .expect("workspace.create should return root pane id") + .to_string(); + + (workspace_id, pane_id) +} + +fn pane_send_input(socket_path: &Path, pane_id: &str, text: &str) { + let request = format!( + "{{\"id\":\"send_input\",\"method\":\"pane.send_input\",\"params\":{{\"pane_id\":\"{pane_id}\",\"text\":\"{}\",\"keys\":[\"Enter\"]}}}}", + text.replace('"', "\\\"") + ); + let response = send_json_request(socket_path, &request); + if response.get("error").is_some() { + panic!("pane.send_input failed: {response}"); + } +} + +fn pane_read_recent(socket_path: &Path, pane_id: &str, lines: usize) -> String { + let response = send_json_request( + socket_path, + &format!( + "{{\"id\":\"pane_read\",\"method\":\"pane.read\",\"params\":{{\"pane_id\":\"{pane_id}\",\"source\":\"recent\",\"lines\":{lines}}}}}" + ), + ); + + if response.get("error").is_some() { + panic!("pane.read failed: {response}"); + } + + response + .pointer("/result/read/text") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() +} + +fn pane_read_recent_contains( + socket_path: &Path, + pane_id: &str, + needle: &str, + timeout: Duration, +) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if pane_read_recent(socket_path, pane_id, 200).contains(needle) { + return true; + } + thread::sleep(Duration::from_millis(50)); + } + false +} + +fn parse_size_after_marker(text: &str, marker: &str) -> Option<(u16, u16)> { + let mut seen_marker = false; + for line in text.lines() { + if !seen_marker { + if line.contains(marker) { + seen_marker = true; + } + continue; + } + + let mut parts = line.split_whitespace(); + let Some(rows_raw) = parts.next() else { + continue; + }; + let Some(cols_raw) = parts.next() else { + continue; + }; + + let Ok(rows) = rows_raw.parse::<u16>() else { + continue; + }; + let Ok(cols) = cols_raw.parse::<u16>() else { + continue; + }; + + return Some((rows, cols)); + } + + None +} + +fn try_read_pane_tty_size( + socket_path: &Path, + pane_id: &str, + timeout: Duration, +) -> Option<(u16, u16)> { + let marker = format!( + "SIZE_MARKER_{}_{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + ); + + pane_send_input(socket_path, pane_id, &format!("echo {marker}; stty size")); + + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let text = pane_read_recent(socket_path, pane_id, 200); + if let Some(size) = parse_size_after_marker(&text, &marker) { + return Some(size); + } + thread::sleep(Duration::from_millis(50)); + } + + None +} + +fn read_pane_tty_size(socket_path: &Path, pane_id: &str, timeout: Duration) -> (u16, u16) { + if let Some(size) = try_read_pane_tty_size(socket_path, pane_id, timeout) { + return size; + } + + let snapshot = pane_read_recent(socket_path, pane_id, 200); + panic!( + "did not observe tty size after marker. pane output:\n{}", + snapshot + ); +} + +// --------------------------------------------------------------------------- +// Minimal bincode v2 varint helpers for protocol tests +// --------------------------------------------------------------------------- + +fn encode_varint_u32(v: u32) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else if v < 65536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +fn encode_varint_u16(v: u16) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> { + let mut buf = encode_varint_u32(variant_idx); + for field in fields { + buf.extend_from_slice(field); + } + buf +} + +fn frame_message(payload: &[u8]) -> Vec<u8> { + let len = payload.len() as u32; + let mut framed = len.to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first_byte}")), + } +} + +fn is_timeout(err: &io::Error) -> bool { + matches!( + err.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + ) +} + +fn read_server_variant(stream: &mut UnixStream, timeout: Duration) -> io::Result<u32> { + stream.set_read_timeout(Some(timeout))?; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + if len == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-length payload", + )); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + + let (variant, _consumed) = decode_varint_u32(&payload, 0) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + Ok(variant) +} + +fn client_handshake( + stream: &mut UnixStream, + version: u32, + cols: u16, + rows: u16, +) -> Result<(), String> { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + + // ClientMessage::Hello = variant 0 + let hello_payload = encode_varint_enum( + 0, + &[ + &encode_varint_u32(version), + &encode_varint_u16(cols), + &encode_varint_u16(rows), + ], + ); + stream + .write_all(&frame_message(&hello_payload)) + .map_err(|e| e.to_string())?; + stream.flush().map_err(|e| e.to_string())?; + + // Read ServerMessage::Welcome = variant 0 + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; + let len = u32::from_le_bytes(len_buf) as usize; + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).map_err(|e| e.to_string())?; + + let mut offset = 0; + let (variant, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + if variant != 0 { + return Err(format!("expected Welcome variant 0, got {variant}")); + } + + let (_server_version, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + + if offset >= payload.len() { + return Err("payload too short for Welcome.error option tag".into()); + } + let option_tag = payload[offset]; + offset += 1; + + if option_tag == 1 { + let (str_len, consumed) = decode_varint_u32(&payload, offset)?; + offset += consumed; + let str_len = str_len as usize; + if offset + str_len > payload.len() { + return Err("payload too short for welcome error string".into()); + } + let err = String::from_utf8(payload[offset..offset + str_len].to_vec()) + .map_err(|e| e.to_string())?; + return Err(format!("handshake rejected: {err}")); + } + + Ok(()) +} + +fn connect_raw_client(client_socket: &Path, cols: u16, rows: u16) -> UnixStream { + let mut stream = UnixStream::connect(client_socket).expect("should connect to client socket"); + client_handshake(&mut stream, 1, cols, rows).expect("handshake should succeed"); + stream +} + +fn send_client_input(stream: &mut UnixStream, data: &[u8]) { + // ClientMessage::Input = variant 1 + let payload = { + let mut buf = encode_varint_u32(1); + buf.extend_from_slice(&encode_varint_u32(data.len() as u32)); + buf.extend_from_slice(data); + buf + }; + stream.write_all(&frame_message(&payload)).unwrap(); + stream.flush().unwrap(); +} + +fn send_client_detach(stream: &mut UnixStream) { + // ClientMessage::Detach = variant 3 + let payload = encode_varint_u32(3); + stream.write_all(&frame_message(&payload)).unwrap(); + stream.flush().unwrap(); +} + +#[derive(Debug, Deserialize)] +struct FrameWire { + cells: Vec<CellWire>, + width: u16, + height: u16, + cursor: Option<CursorWire>, +} + +#[derive(Debug, Deserialize)] +struct CellWire { + symbol: String, + fg: u32, + bg: u32, + modifier: u16, + skip: bool, +} + +#[derive(Debug, Deserialize)] +struct CursorWire { + x: u16, + y: u16, + visible: bool, +} + +fn decode_frame_payload(payload: &[u8]) -> io::Result<FrameWire> { + bincode::serde::decode_from_slice(payload, bincode::config::standard()) + .map(|(frame, consumed)| (frame, consumed)) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string())) + .and_then(|(frame, consumed): (FrameWire, usize)| { + if consumed != payload.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "frame payload had trailing bytes: consumed={}, len={}", + consumed, + payload.len() + ), + )); + } + Ok(frame) + }) +} + +fn read_server_message_payload( + stream: &mut UnixStream, + timeout: Duration, +) -> io::Result<(u32, Vec<u8>)> { + stream.set_read_timeout(Some(timeout))?; + + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + if len == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "zero-length payload", + )); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + + let (variant, consumed) = decode_varint_u32(&payload, 0) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + + Ok((variant, payload[consumed..].to_vec())) +} + +fn drain_server_messages(stream: &mut UnixStream, max_drain: Duration) { + let deadline = Instant::now() + max_drain; + while Instant::now() < deadline { + match read_server_variant(stream, Duration::from_millis(50)) { + Ok(_) => {} + Err(err) if is_timeout(&err) => break, + Err(_) => break, + } + } +} + +fn wait_for_frame(stream: &mut UnixStream, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let remaining = deadline.saturating_duration_since(Instant::now()); + let slice = remaining.min(Duration::from_millis(75)); + match read_server_variant(stream, slice) { + Ok(1) => return true, // ServerMessage::Frame + Ok(_) => {} + Err(err) if is_timeout(&err) => {} + Err(_) => return false, + } + } + false +} + +fn wait_for_frame_matching( + stream: &mut UnixStream, + timeout: Duration, + predicate: impl Fn(&FrameWire) -> bool, +) -> io::Result<bool> { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + let slice = deadline + .saturating_duration_since(Instant::now()) + .min(Duration::from_millis(80)); + match read_server_message_payload(stream, slice) { + Ok((1, frame_payload)) => { + let frame = decode_frame_payload(&frame_payload)?; + if predicate(&frame) { + return Ok(true); + } + } + Ok((_variant, _payload)) => {} + Err(err) if is_timeout(&err) => {} + Err(err) => return Err(err), + } + } + + Ok(false) +} + +fn frame_contains_text(frame: &FrameWire, needle: &str) -> bool { + if frame.cells.is_empty() { + return false; + } + + let row_width = frame.width.max(1) as usize; + let mut full_text = String::new(); + + for row in frame.cells.chunks(row_width) { + for cell in row { + let _ = (cell.fg, cell.bg, cell.modifier, cell.skip); + full_text.push_str(&cell.symbol); + } + full_text.push('\n'); + } + + let _ = frame.height; + if let Some(cursor) = frame.cursor.as_ref() { + let _ = (cursor.x, cursor.y, cursor.visible); + } + + full_text.contains(needle) +} + +#[test] +fn multi_client_allows_multiple_simultaneous_connections() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let mut client_a = connect_raw_client(&client_socket, 120, 40); + let mut client_b = connect_raw_client(&client_socket, 100, 30); + + assert!( + wait_for_frame(&mut client_a, Duration::from_secs(2)), + "client A should receive frames" + ); + assert!( + wait_for_frame(&mut client_b, Duration::from_secs(2)), + "client B should receive frames" + ); + + let ping = ping_socket(&api_socket); + assert!( + ping.contains("pong"), + "server should remain responsive: {ping}" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_effective_size_shrinks_when_smaller_client_joins() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let (_workspace_id, pane_id) = create_workspace_and_root_pane(&api_socket, "size-shrink"); + + let mut large = connect_raw_client(&client_socket, 120, 40); + assert!(wait_for_frame(&mut large, Duration::from_secs(2))); + let large_only_size = read_pane_tty_size(&api_socket, &pane_id, Duration::from_secs(5)); + + let mut small = connect_raw_client(&client_socket, 80, 24); + assert!(wait_for_frame(&mut small, Duration::from_secs(2))); + let with_small_size = read_pane_tty_size(&api_socket, &pane_id, Duration::from_secs(5)); + + assert!( + with_small_size.0 < large_only_size.0 && with_small_size.1 < large_only_size.1, + "effective pane size should shrink when smaller client joins: before={:?}, after={:?}", + large_only_size, + with_small_size + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_broadcasts_frame_updates_to_all_clients_within_500ms() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let mut client_a = connect_raw_client(&client_socket, 100, 30); + let mut client_b = connect_raw_client(&client_socket, 100, 30); + + // Ensure we have an active pane that can reflect input changes. + let (_workspace_id, pane_id) = + create_workspace_and_root_pane(&api_socket, "broadcast-client-a-to-b"); + + // Drain initial frames so we measure the frame caused by new input. + drain_server_messages(&mut client_a, Duration::from_millis(300)); + drain_server_messages(&mut client_b, Duration::from_millis(300)); + + let marker = format!( + "MB{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0) + ); + + let start = Instant::now(); + send_client_input(&mut client_a, marker.as_bytes()); + let received = wait_for_frame_matching(&mut client_b, Duration::from_millis(500), |frame| { + frame_contains_text(frame, &marker) + }) + .expect("frame decoding should succeed"); + + assert!( + received, + "client B should receive a broadcast frame containing client A marker within 500ms" + ); + assert!( + start.elapsed() <= Duration::from_millis(500), + "broadcast frame containing marker should arrive within 500ms, got {:?}", + start.elapsed() + ); + assert!( + pane_read_recent_contains(&api_socket, &pane_id, &marker, Duration::from_secs(5)), + "pane output should include client A marker so broadcast reflects a real state change" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_disconnect_recalculates_to_next_smallest() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let (_workspace_id, pane_id) = + create_workspace_and_root_pane(&api_socket, "size-next-smallest"); + + let mut c120 = connect_raw_client(&client_socket, 120, 40); + let mut c100 = connect_raw_client(&client_socket, 100, 30); + let mut c80 = connect_raw_client(&client_socket, 80, 24); + + assert!(wait_for_frame(&mut c120, Duration::from_secs(2))); + assert!(wait_for_frame(&mut c100, Duration::from_secs(2))); + assert!(wait_for_frame(&mut c80, Duration::from_secs(2))); + + let size_with_three = read_pane_tty_size(&api_socket, &pane_id, Duration::from_secs(5)); + + // Smallest client disconnects; effective size should increase to the next-smallest. + send_client_detach(&mut c80); + drop(c80); + + let deadline = Instant::now() + Duration::from_secs(8); + let mut size_after_smallest_disconnect = None; + while Instant::now() < deadline { + let maybe_size = try_read_pane_tty_size(&api_socket, &pane_id, Duration::from_millis(400)); + if let Some(size) = maybe_size { + if size.0 > size_with_three.0 && size.1 > size_with_three.1 { + size_after_smallest_disconnect = Some(size); + break; + } + } + thread::sleep(Duration::from_millis(60)); + } + + assert!( + size_after_smallest_disconnect.is_some(), + "effective pane size should increase after smallest disconnects: before={:?}, last_seen={:?}", + size_with_three, + try_read_pane_tty_size(&api_socket, &pane_id, Duration::from_millis(300)) + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_smallest_leaving_resizes_up_for_remaining_clients() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let (_workspace_id, pane_id) = create_workspace_and_root_pane(&api_socket, "size-resize-up"); + + let mut large = connect_raw_client(&client_socket, 120, 40); + let mut small = connect_raw_client(&client_socket, 80, 24); + + assert!(wait_for_frame(&mut large, Duration::from_secs(2))); + assert!(wait_for_frame(&mut small, Duration::from_secs(2))); + + let size_with_small_client = read_pane_tty_size(&api_socket, &pane_id, Duration::from_secs(5)); + + drain_server_messages(&mut large, Duration::from_millis(250)); + + send_client_detach(&mut small); + drop(small); + + // Remaining client should receive a new (larger) frame. + assert!( + wait_for_frame(&mut large, Duration::from_secs(2)), + "remaining client should receive resized-up frame" + ); + + let size_after_small_leaves = read_pane_tty_size(&api_socket, &pane_id, Duration::from_secs(5)); + + assert!( + size_after_small_leaves.0 > size_with_small_client.0 + && size_after_small_leaves.1 > size_with_small_client.1, + "remaining clients should get larger effective pane size after smallest leaves: before={:?}, after={:?}", + size_with_small_client, + size_after_small_leaves + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_client_crash_sigkill_does_not_affect_server() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + let mut survivor = connect_raw_client(&client_socket, 100, 30); + assert!(wait_for_frame(&mut survivor, Duration::from_secs(2))); + + let log_path = server_log_path(&config_home); + let connected_before = count_log_occurrences(&log_path, "client connected"); + + let crashing_client = spawn_client_process(&config_home, &runtime_dir, &api_socket); + + let attached_before_kill = wait_for_log_occurrence_count( + &log_path, + "client connected", + connected_before + 1, + Duration::from_secs(8), + ); + assert!( + attached_before_kill, + "thin client must complete handshake/attachment before SIGKILL" + ); + + if let Some(pid) = crashing_client.child.process_id() { + unsafe { + libc::kill(pid as libc::pid_t, libc::SIGKILL); + } + } + let mut crashing_client = crashing_client; + wait_for_child_exit(&mut crashing_client.child); + + let ping = ping_socket(&api_socket); + assert!( + ping.contains("pong"), + "server should stay healthy after SIGKILLed client: {ping}" + ); + + drain_server_messages(&mut survivor, Duration::from_millis(250)); + send_client_input(&mut survivor, b"echo survivor-still-works\n"); + assert!( + wait_for_frame(&mut survivor, Duration::from_secs(2)), + "remaining client should continue receiving frames" + ); + + cleanup_spawned_herdr(server, base); +} + +#[test] +fn multi_client_rapid_connect_disconnect_stress_10_cycles() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let server = spawn_server(&config_home, &runtime_dir, &api_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + for i in 0..10u16 { + let mut client = connect_raw_client(&client_socket, 80 + i, 24 + (i % 4)); + let _ = wait_for_frame(&mut client, Duration::from_millis(500)); + send_client_detach(&mut client); + drop(client); + thread::sleep(Duration::from_millis(40)); + } + + let ping = ping_socket(&api_socket); + assert!( + ping.contains("pong"), + "server should remain healthy after rapid connect/disconnect: {ping}" + ); + + let mut final_client = connect_raw_client(&client_socket, 100, 30); + assert!( + wait_for_frame(&mut final_client, Duration::from_secs(2)), + "new client should still connect and receive frames after stress" + ); + + cleanup_spawned_herdr(server, base); +} diff --git a/tests/server_headless.rs b/tests/server_headless.rs new file mode 100644 index 00000000..98ca0b08 --- /dev/null +++ b/tests/server_headless.rs @@ -0,0 +1,713 @@ +//! Integration tests for headless server mode. + +mod support; + +use std::fs; +use std::io::{BufRead, BufReader, Read, Write}; +use std::os::unix::fs::FileTypeExt; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard, OnceLock}; +use std::thread; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; +use support::{ + cleanup_test_base, register_runtime_dir, register_spawned_herdr_pid, + unregister_spawned_herdr_pid, +}; + +fn unique_test_dir() -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + PathBuf::from(format!( + "/tmp/herdr-server-test-{}-{nanos}", + std::process::id() + )) +} + +struct SpawnedHerdr { + _master: Box<dyn MasterPty + Send>, + child: Box<dyn Child + Send + Sync>, +} + +impl Drop for SpawnedHerdr { + fn drop(&mut self) { + let pid = self.child.process_id(); + let _ = self.child.kill(); + + if let Some(pid) = pid { + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline { + let mut status = 0; + let result = + unsafe { libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG) }; + if result == pid as libc::pid_t || result == -1 { + break; + } + thread::sleep(Duration::from_millis(20)); + } + + unregister_spawned_herdr_pid(Some(pid)); + } + } +} + +fn cleanup_spawned_herdr(spawned: SpawnedHerdr, base: PathBuf) { + drop(spawned); + cleanup_test_base(&base); +} + +fn test_lock() -> MutexGuard<'static, ()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn wait_for_socket(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() && UnixStream::connect(path).is_ok() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("socket did not appear at {}", path.display()); +} + +fn wait_for_file(path: &Path, timeout: Duration) { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if path.exists() { + return; + } + thread::sleep(Duration::from_millis(25)); + } + panic!("file did not appear at {}", path.display()); +} + +fn spawn_server( + config_home: &Path, + runtime_dir: &Path, + api_socket_path: &Path, + _client_socket_path: &Path, +) -> SpawnedHerdr { + fs::create_dir_all(config_home.join("herdr")).unwrap(); + fs::create_dir_all(runtime_dir).unwrap(); + register_runtime_dir(runtime_dir); + fs::write( + config_home.join("herdr/config.toml"), + "onboarding = false\n", + ) + .unwrap(); + + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", config_home); + cmd.env("XDG_RUNTIME_DIR", runtime_dir); + cmd.env("HERDR_SOCKET_PATH", api_socket_path); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let child = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child.process_id()); + drop(pair.slave); + + SpawnedHerdr { + _master: pair.master, + child, + } +} + +fn ping_socket(socket_path: &Path) -> String { + let mut stream = UnixStream::connect(socket_path).expect("should connect to API socket"); + + let request = r#"{"id":"1","method":"ping","params":{}}"#; + writeln!(stream, "{}", request).unwrap(); + + let mut reader = BufReader::new(stream); + let mut response = String::new(); + reader.read_line(&mut response).unwrap(); + response.trim().to_string() +} + +/// Sends a Hello message over the client socket and reads the Welcome response. +/// Uses bincode v2 wire format: [u32LE length][bincode payload] +/// bincode v2 standard config uses VarintEncoding: +/// - Integers < 251 are encoded as a single byte +/// - Enum variant index is encoded as u32 varint +/// - Option discriminant is always a single byte (0=None, 1=Some) +/// - String: length (varint) + UTF-8 bytes +fn client_handshake( + stream: &mut UnixStream, + version: u32, + cols: u16, + rows: u16, +) -> Result<(u32, Option<String>), String> { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .map_err(|e| e.to_string())?; + + // Encode Hello message using bincode v2 varint format. + // ClientMessage::Hello is variant 0. + let hello_payload = encode_varint_enum( + 0, + &[ + &encode_varint_u32(version), + &encode_varint_u16(cols), + &encode_varint_u16(rows), + ], + ); + let framed = frame_message(&hello_payload); + stream.write_all(&framed).map_err(|e| e.to_string())?; + stream.flush().map_err(|e| e.to_string())?; + + // Read the framed response. + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).map_err(|e| e.to_string())?; + let len = u32::from_le_bytes(len_buf) as usize; + + if len > 2 * 1024 * 1024 { + return Err(format!("oversized response: {len}")); + } + + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).map_err(|e| e.to_string())?; + + // Decode Welcome: ServerMessage variant 0 = Welcome { version: u32, error: Option<String> } + decode_welcome(&payload) +} + +/// Encode a varint u32 value according to bincode v2 VarintEncoding. +fn encode_varint_u32(v: u32) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else if v < 65536 { + let mut buf = vec![251u8]; + buf.extend_from_slice(&(v as u16).to_le_bytes()); + buf + } else { + let mut buf = vec![252u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode a varint u16 value. +fn encode_varint_u16(v: u16) -> Vec<u8> { + if v < 251 { + vec![v as u8] + } else { + let mut buf = vec![251u8]; + buf.extend_from_slice(&v.to_le_bytes()); + buf + } +} + +/// Encode an enum variant with its fields. +fn encode_varint_enum(variant_idx: u32, fields: &[&[u8]]) -> Vec<u8> { + let mut buf = encode_varint_u32(variant_idx); + for field in fields { + buf.extend_from_slice(field); + } + buf +} + +/// Frame a message with u32LE length prefix. +fn frame_message(payload: &[u8]) -> Vec<u8> { + let len = payload.len() as u32; + let mut framed = len.to_le_bytes().to_vec(); + framed.extend_from_slice(payload); + framed +} + +/// Decode a varint u32 from a byte slice at the given offset. +/// Returns (value, bytes_consumed). +fn decode_varint_u32(payload: &[u8], offset: usize) -> Result<(u32, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u32, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v as u32, 3)) + } + 252 => { + if offset + 5 > payload.len() { + return Err("payload too short for u32 varint".into()); + } + let v = u32::from_le_bytes( + payload[offset + 1..offset + 5] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 5)) + } + _ => Err(format!("unsupported varint tag: {first_byte}")), + } +} + +/// Decode a varint u16 from a byte slice at the given offset. +#[allow(dead_code)] +fn decode_varint_u16(payload: &[u8], offset: usize) -> Result<(u16, usize), String> { + if offset >= payload.len() { + return Err("payload too short for varint".into()); + } + let first_byte = payload[offset]; + match first_byte { + 0..=250 => Ok((first_byte as u16, 1)), + 251 => { + if offset + 3 > payload.len() { + return Err("payload too short for u16 varint".into()); + } + let v = u16::from_le_bytes( + payload[offset + 1..offset + 3] + .try_into() + .map_err(|e: std::array::TryFromSliceError| e.to_string())?, + ); + Ok((v, 3)) + } + _ => Err(format!("unsupported varint tag for u16: {first_byte}")), + } +} + +/// Decode a ServerMessage::Welcome from bincode v2 payload. +fn decode_welcome(payload: &[u8]) -> Result<(u32, Option<String>), String> { + let mut offset = 0; + + // Variant index (should be 0 for Welcome) + let (variant, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + if variant != 0 { + return Err(format!( + "expected Welcome (variant 0), got variant {variant}" + )); + } + + // version: u32 + let (version, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + + // error: Option<String> — discriminant is always 1 byte + if offset >= payload.len() { + return Err("payload too short for Option tag".into()); + } + let option_tag = payload[offset]; + offset += 1; + + let error = if option_tag == 1 { + // Some(String) — length as varint + UTF-8 bytes + let (str_len, consumed) = decode_varint_u32(payload, offset)?; + offset += consumed; + let str_len = str_len as usize; + + if offset + str_len > payload.len() { + return Err("payload too short for string content".into()); + } + let s = String::from_utf8(payload[offset..offset + str_len].to_vec()) + .map_err(|e| e.to_string())?; + Some(s) + } else { + None + }; + + Ok((version, error)) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[test] +fn server_creates_both_sockets() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + + // Wait for both sockets to appear. + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Verify the client socket is a socket file. + let metadata = fs::metadata(&client_socket).unwrap(); + let file_type = metadata.file_type(); + assert!( + file_type.is_socket(), + "client socket should be a socket file" + ); + + // Verify the API socket works. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "ping should return pong: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_starts_without_terminal() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + + // Wait for the API socket to appear — proves the server started. + wait_for_socket(&api_socket, Duration::from_secs(10)); + + // The server process should be running. + if let Some(pid) = spawned.child.process_id() { + let result = unsafe { libc::kill(pid as libc::pid_t, 0) }; + assert_eq!(result, 0, "server process should be running"); + } + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_api_responds_to_ping() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + + // Ping the API socket. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should respond to ping: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_removes_client_socket_on_exit() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let mut spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Kill the server. + let _ = spawned.child.kill(); + let _ = spawned.child.wait(); + + // Give it a moment to clean up. + thread::sleep(Duration::from_millis(300)); + + // The client socket should be removed (best-effort by Drop). + // If it still exists, it should be stale (not connectable). + if client_socket.exists() { + assert!( + UnixStream::connect(&client_socket).is_err(), + "stale client socket should not accept connections" + ); + } + + drop(spawned); + cleanup_test_base(&base); +} + +#[test] +fn server_cleans_up_stale_client_socket() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Create a stale client socket file (simulating a crashed server). + fs::create_dir_all(&runtime_dir).unwrap(); + register_runtime_dir(&runtime_dir); + { + let _listener = std::os::unix::net::UnixListener::bind(&client_socket).unwrap(); + // Drop the listener so the socket becomes stale. + } + + // Now start the server — it should clean up the stale socket. + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + + // The API should work. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should respond to ping: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn server_persists_after_client_disconnect() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect to the client socket and then immediately disconnect. + { + let _stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + // Immediately drop the connection. + } + + // Give the server a moment to process the disconnect. + thread::sleep(Duration::from_millis(200)); + + // The server should still be running and the API should still respond. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "API should still respond after client disconnect: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn duplicate_server_start_fails_gracefully() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + // Start the first server. + let spawned1 = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + + // Try to start a second server — it should fail. + let pair = native_pty_system() + .openpty(PtySize { + rows: 24, + cols: 80, + pixel_width: 0, + pixel_height: 0, + }) + .unwrap(); + + let mut cmd = CommandBuilder::new(env!("CARGO_BIN_EXE_herdr")); + cmd.arg("server"); + cmd.env("XDG_CONFIG_HOME", &config_home); + cmd.env("XDG_RUNTIME_DIR", &runtime_dir); + cmd.env("HERDR_SOCKET_PATH", &api_socket); + cmd.env_remove("HERDR_CLIENT_SOCKET_PATH"); + cmd.env("SHELL", "/bin/sh"); + cmd.env_remove("HERDR_ENV"); + + let mut child2 = pair.slave.spawn_command(cmd).unwrap(); + register_spawned_herdr_pid(child2.process_id()); + drop(pair.slave); + + // Wait for the second server to exit. + let exit_status = child2.wait().unwrap(); + unregister_spawned_herdr_pid(child2.process_id()); + + // The second server should exit with a non-zero code. + assert!(!exit_status.success(), "duplicate server start should fail"); + + cleanup_spawned_herdr(spawned1, base); +} + +#[test] +fn client_handshake_succeeds() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect to the client socket and perform a handshake. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + + // Send Hello with version 1, 80 cols, 24 rows. + let (version, error) = + client_handshake(&mut stream, 1, 80, 24).expect("handshake should succeed"); + + assert_eq!(version, 1, "server should report protocol version 1"); + assert!( + error.is_none(), + "handshake should not have an error: {:?}", + error + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn client_handshake_rejects_incompatible_version() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect to the client socket and send Hello with version 0 (pre-persistence). + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + + let (version, error) = client_handshake(&mut stream, 0, 80, 24) + .expect("should read Welcome response even on rejection"); + + assert_eq!(version, 1, "server should report its version 1"); + assert!( + error.is_some(), + "version 0 should be rejected with an error" + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn client_handshake_clamps_small_terminal_size() { + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Send Hello with 0x0 terminal size — should be clamped. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + + let (version, error) = client_handshake(&mut stream, 1, 0, 0) + .expect("handshake with 0x0 should succeed (server clamps)"); + + assert_eq!(version, 1); + assert!( + error.is_none(), + "0x0 size should be accepted (clamped): {:?}", + error + ); + + cleanup_spawned_herdr(spawned, base); +} + +#[test] +fn no_hello_client_closed_within_five_seconds() { + // Client connection that sends no Hello is closed within 5 seconds. + // The server sets a handshake timeout of 4 seconds to guarantee the connection + // is closed within the 5-second deadline even with OS overhead. + let _lock = test_lock(); + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let api_socket = runtime_dir.join("herdr.sock"); + let client_socket = runtime_dir.join("herdr-client.sock"); + + let spawned = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); + wait_for_socket(&api_socket, Duration::from_secs(10)); + wait_for_file(&client_socket, Duration::from_secs(10)); + + // Connect but don't send Hello — just a raw connection. + let mut stream = UnixStream::connect(&client_socket).expect("should connect to client socket"); + + // Set a read timeout longer than the handshake timeout so we can detect + // when the server closes the connection. + stream + .set_read_timeout(Some(Duration::from_secs(6))) + .unwrap(); + + let start = Instant::now(); + + // Try to read from the stream. The server should close the connection + // within 5 seconds, causing our read to return with an error (EOF or + // connection reset). + let mut buf = [0u8; 1024]; + let result = stream.read(&mut buf); + let elapsed = start.elapsed(); + + // The read should fail (connection closed by server). + assert!( + result.is_err() || result.unwrap() == 0, + "server should close the connection when no Hello is sent" + ); + + // The connection should be closed within 5 seconds. + assert!( + elapsed < Duration::from_secs(5), + "connection should be closed within 5 seconds, took {:?}", + elapsed + ); + + // Verify the server is still healthy — a proper client can still connect. + let mut good_stream = + UnixStream::connect(&client_socket).expect("should connect after no-hello client"); + let (version, error) = client_handshake(&mut good_stream, 1, 80, 24) + .expect("proper handshake should still work after no-hello client"); + assert_eq!(version, 1); + assert!(error.is_none()); + + // API should still work. + let response = ping_socket(&api_socket); + assert!( + response.contains("pong"), + "server should still respond to ping: {response}" + ); + + cleanup_spawned_herdr(spawned, base); +} diff --git a/tests/support/mod.rs b/tests/support/mod.rs new file mode 100644 index 00000000..567db154 --- /dev/null +++ b/tests/support/mod.rs @@ -0,0 +1,418 @@ +use std::collections::HashSet; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, Once, OnceLock}; +use std::thread; +use std::time::{Duration, Instant}; + +static PID_REGISTRY: OnceLock<Mutex<HashSet<u32>>> = OnceLock::new(); +static RUNTIME_DIR_REGISTRY: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new(); +static INIT: Once = Once::new(); +static CLEANUP_GUARD: OnceLock<CleanupGuard> = OnceLock::new(); +const WATCHDOG_SCAN_INTERVAL: Duration = Duration::from_secs(1); +const RUNTIME_OWNER_MARKER: &str = ".herdr-test-owner-pid"; + +pub fn register_spawned_herdr_pid(pid: Option<u32>) { + let Some(pid) = pid else { + return; + }; + + ensure_cleanup_hooks(); + let mut registry = pid_registry_lock(); + registry.insert(pid); +} + +pub fn unregister_spawned_herdr_pid(pid: Option<u32>) { + let Some(pid) = pid else { + return; + }; + + if let Some(registry) = PID_REGISTRY.get() { + let mut guard = registry + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + guard.remove(&pid); + } +} + +pub fn register_runtime_dir(path: &Path) { + ensure_cleanup_hooks(); + + let _ = fs::create_dir_all(path); + let _ = fs::write( + path.join(RUNTIME_OWNER_MARKER), + std::process::id().to_string(), + ); + + let mut runtime_dirs = runtime_dir_registry_lock(); + runtime_dirs.insert(path.to_path_buf()); +} + +pub fn unregister_runtime_dir(path: &Path) { + if let Some(registry) = RUNTIME_DIR_REGISTRY.get() { + let mut guard = registry + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + guard.remove(path); + } +} + +pub fn cleanup_test_base(base: &Path) { + let runtime_dir = base.join("runtime"); + let runtime_dirs = HashSet::from([runtime_dir.clone()]); + + terminate_servers_for_runtime_dirs(&runtime_dirs); + unregister_runtime_dir(&runtime_dir); + let _ = fs::remove_dir_all(base); +} + +pub fn cleanup_registered_herdr_pids() { + let pids: Vec<u32> = { + let mut registry = pid_registry_lock(); + registry.drain().collect() + }; + + for pid in pids { + terminate_pid(pid); + } + + let runtime_dirs: HashSet<PathBuf> = { + let mut runtime_dirs = runtime_dir_registry_lock(); + runtime_dirs.drain().collect() + }; + + terminate_servers_for_runtime_dirs(&runtime_dirs); + let _ = cleanup_servers_with_missing_runtime_dir(); +} + +fn ensure_cleanup_hooks() { + INIT.call_once(|| { + let _ = cleanup_servers_with_missing_runtime_dir(); + start_global_watchdog(); + + let _ = CLEANUP_GUARD.set(CleanupGuard); + + let previous_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |panic_info| { + cleanup_registered_herdr_pids(); + previous_hook(panic_info); + })); + + let _ = ctrlc::set_handler(|| { + cleanup_registered_herdr_pids(); + std::process::exit(130); + }); + + unsafe { + libc::atexit(run_atexit_cleanup); + } + }); +} + +fn pid_registry_lock() -> std::sync::MutexGuard<'static, HashSet<u32>> { + PID_REGISTRY + .get_or_init(|| Mutex::new(HashSet::new())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn runtime_dir_registry_lock() -> std::sync::MutexGuard<'static, HashSet<PathBuf>> { + RUNTIME_DIR_REGISTRY + .get_or_init(|| Mutex::new(HashSet::new())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +fn registered_runtime_dirs_snapshot() -> HashSet<PathBuf> { + if let Some(runtime_dirs) = RUNTIME_DIR_REGISTRY.get() { + runtime_dirs + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } else { + HashSet::new() + } +} + +fn should_terminate_runtime_dir( + runtime_dir: &Path, + registered_runtime_dirs: &HashSet<PathBuf>, +) -> bool { + if !registered_runtime_dirs.contains(runtime_dir) { + return false; + } + + if !runtime_dir.exists() { + return true; + } + + !runtime_dir_owner_alive(runtime_dir) +} + +fn start_global_watchdog() { + thread::spawn(|| loop { + thread::sleep(WATCHDOG_SCAN_INTERVAL); + + if let Err(err) = cleanup_servers_with_missing_runtime_dir() { + eprintln!("herdr test cleanup watchdog error: {err}"); + } + }); +} + +fn cleanup_servers_with_missing_runtime_dir() -> std::io::Result<()> { + let registered_runtime_dirs = registered_runtime_dirs_snapshot(); + if registered_runtime_dirs.is_empty() { + return Ok(()); + } + + for pid in iter_worktree_server_pids()? { + let Some(runtime_dir) = process_runtime_dir(pid)? else { + continue; + }; + + if should_terminate_runtime_dir(&runtime_dir, ®istered_runtime_dirs) { + terminate_pid(pid); + } + } + + Ok(()) +} + +fn terminate_servers_for_runtime_dirs(runtime_dirs: &HashSet<PathBuf>) { + if runtime_dirs.is_empty() { + return; + } + + let Ok(pids) = iter_worktree_server_pids() else { + return; + }; + + for pid in pids { + let Ok(runtime_dir) = process_runtime_dir(pid) else { + continue; + }; + + let Some(runtime_dir) = runtime_dir else { + continue; + }; + + if runtime_dirs.contains(&runtime_dir) { + terminate_pid(pid); + } + } +} + +fn iter_worktree_server_pids() -> std::io::Result<Vec<u32>> { + let own_pid = std::process::id(); + let mut pids = Vec::new(); + + for entry in fs::read_dir("/proc")? { + let entry = entry?; + let file_name = entry.file_name(); + let Some(pid) = file_name.to_str().and_then(|name| name.parse::<u32>().ok()) else { + continue; + }; + + if pid == own_pid { + continue; + } + + if is_worktree_herdr_server_process(pid) { + pids.push(pid); + } + } + + Ok(pids) +} + +fn is_worktree_herdr_server_process(pid: u32) -> bool { + let Some(exe_path) = proc_link_target(pid, "exe") else { + return false; + }; + + if !is_worktree_herdr_binary(&exe_path) { + return false; + } + + let Ok(cmdline) = read_cmdline(pid) else { + return false; + }; + + cmdline.iter().any(|arg| arg == "server") +} + +fn proc_link_target(pid: u32, link: &str) -> Option<PathBuf> { + fs::read_link(format!("/proc/{pid}/{link}")).ok() +} + +fn read_cmdline(pid: u32) -> std::io::Result<Vec<String>> { + let cmdline = fs::read(format!("/proc/{pid}/cmdline"))?; + Ok(cmdline + .split(|byte| *byte == 0) + .filter(|chunk| !chunk.is_empty()) + .map(|chunk| String::from_utf8_lossy(chunk).to_string()) + .collect()) +} + +fn process_runtime_dir(pid: u32) -> std::io::Result<Option<PathBuf>> { + let environ = fs::read(format!("/proc/{pid}/environ"))?; + + let mut socket_path: Option<PathBuf> = None; + + for entry in environ.split(|byte| *byte == 0) { + if entry.is_empty() { + continue; + } + + let kv = String::from_utf8_lossy(entry); + if let Some(value) = kv.strip_prefix("XDG_RUNTIME_DIR=") { + return Ok(Some(PathBuf::from(value))); + } + + if let Some(value) = kv.strip_prefix("HERDR_SOCKET_PATH=") { + socket_path = Some(PathBuf::from(value)); + } + } + + Ok(socket_path.and_then(|path| path.parent().map(Path::to_path_buf))) +} + +fn runtime_dir_owner_alive(runtime_dir: &Path) -> bool { + let marker = runtime_dir.join(RUNTIME_OWNER_MARKER); + let Ok(contents) = fs::read_to_string(marker) else { + return false; + }; + + let Ok(owner_pid) = contents.trim().parse::<libc::pid_t>() else { + return false; + }; + + process_exists(owner_pid) +} + +fn is_worktree_herdr_binary(path: &Path) -> bool { + path.ends_with("target/debug/herdr") + && path + .components() + .any(|component| component.as_os_str() == "herdr-worktrees") +} + +extern "C" fn run_atexit_cleanup() { + cleanup_registered_herdr_pids(); +} + +struct CleanupGuard; + +impl Drop for CleanupGuard { + fn drop(&mut self) { + cleanup_registered_herdr_pids(); + } +} + +fn terminate_pid(pid: u32) { + let pid_t = pid as libc::pid_t; + + if process_exists(pid_t) { + unsafe { + libc::kill(pid_t, libc::SIGTERM); + } + } + + if wait_for_pid_exit(pid_t, Duration::from_millis(400)) { + return; + } + + if process_exists(pid_t) { + unsafe { + libc::kill(pid_t, libc::SIGKILL); + } + } + + let _ = wait_for_pid_exit(pid_t, Duration::from_secs(2)); +} + +fn wait_for_pid_exit(pid: libc::pid_t, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + + while Instant::now() < deadline { + if !process_exists(pid) { + return true; + } + + let mut status = 0; + let result = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + if result == pid { + return true; + } + + if result == -1 { + match std::io::Error::last_os_error().raw_os_error() { + Some(libc::ECHILD) => { + // Not our child (or already reaped elsewhere). Poll /proc existence + // until the process is truly gone. + if !process_exists(pid) { + return true; + } + } + Some(libc::ESRCH) => return true, + _ => { + if !process_exists(pid) { + return true; + } + } + } + } + + thread::sleep(Duration::from_millis(20)); + } + + !process_exists(pid) +} + +fn process_exists(pid: libc::pid_t) -> bool { + let result = unsafe { libc::kill(pid, 0) }; + if result == 0 { + true + } else { + std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn unique_missing_runtime_dir(label: &str) -> PathBuf { + let unique = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + std::env::temp_dir().join(format!( + "herdr-watchdog-scoping-{label}-{}-{unique}", + std::process::id() + )) + } + + #[test] + fn watchdog_scoping_does_not_terminate_missing_unregistered_runtime_dir() { + let runtime_dir = unique_missing_runtime_dir("unregistered"); + let registered_runtime_dirs = HashSet::new(); + + assert!( + !should_terminate_runtime_dir(&runtime_dir, ®istered_runtime_dirs), + "missing runtime dirs must not be killable until they are proven session-owned" + ); + } + + #[test] + fn watchdog_scoping_terminates_missing_registered_runtime_dir() { + let runtime_dir = unique_missing_runtime_dir("registered"); + let mut registered_runtime_dirs = HashSet::new(); + registered_runtime_dirs.insert(runtime_dir.clone()); + + assert!( + should_terminate_runtime_dir(&runtime_dir, ®istered_runtime_dirs), + "missing runtime dirs that are session-owned should be considered killable" + ); + } +} |