feat: add persistent server-client sessions

   Make herdr persistent by default.

   Launching herdr now starts or reattaches to a background session
   server. Clients can detach and reattach while panes and agent
   processes keep running. Session mode now supports multi-client attach,
   auto-detect startup, and a thin-client/headless-server split.

   This also refactors the large app, ui, input, pane, config,
   workspace, and persistence modules into smaller focused submodules,
   while preserving behavior and colocating tests with the code they
   exercise.

   Upgrade notes:
   - persistence mode is now the default
   - in-app quit detaches the current client instead of stopping the server
   - use `herdr server stop` to stop the background session
   - use `--no-session` for the old single-process behavior
   - default socket paths now live under the config directory

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, &registered_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, &registered_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, &registered_runtime_dirs),
+            "missing runtime dirs that are session-owned should be considered killable"
+        );
+    }
+}