// daemon_attach.rs --- `pmacs --daemon-attach` byte bridge (T M5.7b). //! Far-side byte bridge between stdin/stdout and a local daemon //! socket. //! //! `pmacs --daemon-attach` is the mode that runs on the far side of //! an SSH connection (or any other transport that can spawn a //! process and bridge its stdio). It connects to the local daemon's //! Unix socket and forwards bytes verbatim: //! //! ```text //! stdin ──> socket //! socket ──> stdout //! ``` //! //! # Transparent forwarding //! //! The bridge is **not** a protocol participant. It does not parse //! [`crate::protocol::Hello`] frames or speak postcard. It is a pair //! of `std::io::copy` loops. The Hello/AttachRequest handshake //! happens between the *real* frontend (on the user's laptop) and //! the *real* daemon (on the same machine as the bridge); the bridge //! just shuttles the bytes those two parties exchange. //! //! This is the simplest possible thing that works, and it's correct: //! the bridge has no knowledge of protocol versioning, no opinion on //! framing, and adds no surface to the wire format. Adding a new //! protocol message in v0.2 does not require touching this file. //! //! # Lifetime //! //! The bridge process's lifetime equals the attach session's //! lifetime. When the user detaches (F12 → Detach → daemon Goodbye → //! socket close → bridge sees EOF → bridge exits → SSH child exits), //! both copy loops return naturally and `run_daemon_attach` returns //! `Ok(())`. When the SSH client closes its end of the pipe (e.g., //! the user kills the local frontend), the bridge's stdin sees EOF, //! the stdin→socket copy returns, the socket→stdout copy returns //! when the daemon notices and closes its side, and the function //! returns `Ok(())`. //! //! # Auto-start (M5.7c) //! //! If the daemon is not already listening when the bridge tries to //! connect, [`ensure_daemon_running`] spawns `pmacs --daemon //! --socket PATH` as a child process and polls until the socket //! becomes connectable (or a timeout expires). The bridge then //! proceeds with the byte-pump as if the daemon had been there all //! along. //! //! **No setsid.** The crate forbids `unsafe`, so we cannot use //! `Command::pre_exec` to detach the daemon from our session. We //! rely on a different property: in the typical SSH spawn pattern //! (`ssh host pmacs --daemon-attach`), SSH does not allocate a //! controlling terminal, so there is no terminal-hangup channel to //! deliver SIGHUP through. The daemon's parent (the bridge) exits; //! the daemon is re-parented to init/PID 1; no signal is sent. The //! daemon survives across attach sessions, which is the contract. //! Edge cases (e.g., explicit `kill -- -PGID` against the bridge's //! process group) are accepted limitations for v0.1. use std::io::{Read, Write}; use std::net::Shutdown; use std::os::fd::AsFd; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::sync::mpsc; use std::thread; use std::time::{Duration, Instant}; /// Errors produced by the daemon-attach bridge. #[derive(Debug)] pub enum BridgeError { /// Could not connect to the daemon at `socket`, and auto-start /// also failed or timed out. v0.1 messages this with a pointer /// at the manual recovery path. Connect { /// The socket path the bridge tried to connect to. socket: PathBuf, /// The underlying I/O error from `UnixStream::connect`. source: std::io::Error, }, /// Auto-start could not spawn `pmacs --daemon` as a subprocess. /// Distinct from [`Self::Connect`] because the failure happened /// before the daemon had a chance to bind anything. SpawnFailed { /// Path the spawner tried to invoke (typically the running /// pmacs binary, from `current_exe`). exe: PathBuf, /// The I/O error from `Command::spawn`. source: std::io::Error, }, /// Auto-start spawned a daemon process, but the socket did not /// become connectable within the configured timeout. The daemon /// child is left running — it may bind eventually or it may /// have failed at startup; the user can investigate by running /// `pmacs --daemon --socket PATH` directly to see its stderr. AutoStartTimeout { /// The socket path that never became connectable. socket: PathBuf, /// The timeout that elapsed. after: Duration, }, /// Generic I/O error while bridging bytes (clone, shutdown). Io(std::io::Error), } impl std::fmt::Display for BridgeError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Connect { socket, source } => write!( f, "could not connect to daemon at {}: {source}; \ start one first with `pmacs --daemon` and try again", socket.display() ), Self::SpawnFailed { exe, source } => write!( f, "could not spawn daemon ({}): {source}; \ verify the pmacs binary is on PATH and executable", exe.display(), ), Self::AutoStartTimeout { socket, after } => write!( f, "auto-started daemon did not bind {} within {:.1}s; \ run `pmacs --daemon --socket {}` directly to see \ startup errors", socket.display(), after.as_secs_f32(), socket.display(), ), Self::Io(e) => write!(f, "bridge I/O error: {e}"), } } } impl std::error::Error for BridgeError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::Connect { source, .. } | Self::SpawnFailed { source, .. } | Self::Io(source) => { Some(source) } Self::AutoStartTimeout { .. } => None, } } } /// Default timeout for [`ensure_daemon_running`] to wait for an /// auto-started daemon to bind its socket. /// /// The daemon's startup work (lockfile, socket bind, init.lua) is /// fast — sub-second on a healthy machine — but a busy or cold /// system can be slower. Five seconds is generous without forcing /// users to wait long when something is genuinely wrong. const AUTO_START_TIMEOUT: Duration = Duration::from_secs(5); /// Polling interval while waiting for the auto-started daemon's /// socket to become connectable. const AUTO_START_POLL_INTERVAL: Duration = Duration::from_millis(50); /// Shared with `attach.rs`: when set to a non-empty, non-"0" value, /// emit stderr breadcrumbs. The remote bridge's stderr is the SSH /// child's stderr, so this stays off the protocol stdout stream. const PMACS_ATTACH_DEBUG: &str = "PMACS_ATTACH_DEBUG"; const PMACS_ATTACH_PROTOCOL_FD: &str = "PMACS_ATTACH_PROTOCOL_FD"; fn bridge_debug_enabled() -> bool { std::env::var_os(PMACS_ATTACH_DEBUG).is_some_and(|v| !v.is_empty() && v != "0") } fn bridge_debug(msg: impl AsRef) { // If fd 2 carries the protocol stream, stderr diagnostics would // corrupt it. The local attach side deliberately does not enable // remote debug in that mode, but keep this defensive. if bridge_debug_enabled() && protocol_fd() != 2 { eprintln!("pmacs daemon-attach debug: {}", msg.as_ref()); } } fn protocol_fd() -> i32 { match std::env::var(PMACS_ATTACH_PROTOCOL_FD).ok().as_deref() { Some("2") => 2, _ => 1, } } /// Run `pmacs --daemon-attach`: ensure a daemon is listening at /// `socket_path` (auto-starting one if not), then bridge our /// stdin/stdout to it. /// /// This is the entry point the binary calls. The auto-start step /// runs first so [`run_bridge`]'s `Connect` error path is reached /// only when *something* genuinely cannot reach the daemon (spawn /// failed, timeout fired, etc.). // `PathBuf` by value matches the other CLI entry points // (`run_attach`, `run_daemon`); pedantic clippy is wrong here. #[allow(clippy::needless_pass_by_value)] pub fn run_daemon_attach(socket_path: PathBuf) -> Result<(), BridgeError> { bridge_debug(format!( "starting bridge for socket {}", socket_path.display() )); let socket = ensure_daemon_running(&socket_path)?; bridge_debug("daemon socket ready; entering byte bridge"); // Use the owned `Stdin` / `Stdout` (not `.lock()`) so the halves // can move into threads with `'static` lifetimes. Both are // `Send` and behave as expected for line-oriented or byte // streaming use. let stdin = std::io::stdin(); let output = RawProtocolOutput::new(protocol_fd()); run_bridge_connected(socket, stdin, output) } /// Make sure a daemon is listening at `socket_path`, auto-starting /// one if necessary. /// /// 1. Try to connect. If that succeeds, the daemon is already /// running — return that stream for the bridge to use. /// 2. Otherwise spawn `pmacs --daemon --socket PATH` as a child, /// then poll the socket every [`AUTO_START_POLL_INTERVAL`] until /// a connect succeeds or [`AUTO_START_TIMEOUT`] elapses. The /// successful poll stream is returned to the bridge. /// /// On timeout the spawned daemon child is **not** killed — it may /// be slow to start rather than broken. The user can investigate by /// running the daemon directly to see its stderr. pub(crate) fn ensure_daemon_running(socket_path: &Path) -> Result { ensure_daemon_running_with(socket_path, AUTO_START_TIMEOUT, spawn_daemon_subprocess) } /// Spawner-injected variant of [`ensure_daemon_running`] used by /// tests. /// /// `spawner` is invoked with the resolved socket path when the /// initial connect fails; it is responsible for arranging that the /// path becomes connectable. In production, `spawner` is /// [`spawn_daemon_subprocess`]. Tests substitute closures that /// either bind a `UnixListener` from a worker thread (success) or /// do nothing (timeout). fn ensure_daemon_running_with( socket_path: &Path, timeout: Duration, spawner: F, ) -> Result where F: FnOnce(&Path) -> Result<(), BridgeError>, { // Fast path: daemon already listening. Keep this connection and // use it as the bridge socket. The daemon speaks first, so a // disposable connect probe is not transparent: dropping it before // reading the daemon's Hello can make the daemon log BrokenPipe. match UnixStream::connect(socket_path) { Ok(stream) => { bridge_debug(format!( "connected to existing daemon at {}", socket_path.display() )); return Ok(stream); } Err(e) => { bridge_debug(format!( "initial connect to {} failed: {e}; attempting auto-start", socket_path.display() )); } } // Slow path: ask the spawner to arrange a daemon, then poll. spawner(socket_path)?; bridge_debug("auto-start command spawned; polling socket"); let deadline = Instant::now() + timeout; loop { if let Ok(stream) = UnixStream::connect(socket_path) { bridge_debug(format!( "connected to auto-started daemon at {}", socket_path.display() )); return Ok(stream); } if Instant::now() >= deadline { bridge_debug(format!( "auto-start timed out after {:.1}s", timeout.as_secs_f32() )); return Err(BridgeError::AutoStartTimeout { socket: socket_path.to_path_buf(), after: timeout, }); } thread::sleep(AUTO_START_POLL_INTERVAL); } } /// Production spawner: invoke the running pmacs binary with /// `--daemon --socket PATH`. /// /// All three standard streams are redirected to `/dev/null` — /// stdin because the daemon doesn't read it, stdout/stderr because /// our own stdout is reserved for the bridge's protocol traffic and /// any daemon log noise would corrupt the wire. Users diagnosing /// startup errors run `pmacs --daemon` directly, where stderr is /// inherited normally. fn spawn_daemon_subprocess(socket_path: &Path) -> Result<(), BridgeError> { let exe = std::env::current_exe().map_err(|source| BridgeError::SpawnFailed { exe: PathBuf::from(""), source, })?; bridge_debug(format!( "spawning daemon subprocess: {} --daemon --socket {}", exe.display(), socket_path.display() )); Command::new(&exe) .arg("--daemon") .arg("--socket") .arg(socket_path) .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::null()) .spawn() .map_err(|source| BridgeError::SpawnFailed { exe: exe.clone(), source, }) .map(|child| { bridge_debug(format!("spawned daemon subprocess pid {}", child.id())); })?; Ok(()) } /// Generic byte-bridge implementation, parameterized over the local /// reader and writer for testability. /// /// Spawns a thread for the local-input → socket direction and runs /// the socket → local-output direction on the calling thread. When /// the caller-side copy returns (clean EOF or error), the socket is /// shutdown to wake the stdin thread's next write, and we attempt /// to join. /// /// # Tear-down /// /// Two natural exits drive the bridge's lifetime: /// * **Local input EOF** — caller closed stdin (in production: SSH /// client disconnected). The stdin thread's `copy` returns; the /// thread exits. The main `socket → output` copy continues until /// the daemon notices and closes the socket (typically after the /// detach round-trip). /// * **Socket EOF** — daemon closed the connection. The main /// `socket → output` copy returns immediately. We shutdown the /// socket to wake any pending write on the stdin thread, then /// join. /// /// In both paths, `run_bridge` returns `Ok(())`. The caller is the /// `pmacs --daemon-attach` CLI process whose lifetime equals the /// bridge's; on return, the process exits and the OS reaps any /// stdin thread still blocked in a read of stdin (production-only; /// tests close their input writers explicitly). /// `std::io::copy` plus a `flush` after every chunk. /// /// Postcard frames carry no newlines, but the production /// `local_output` for the bridge is `std::io::Stdout`, whose /// `LineWriter` only flushes on `\n` or buffer-fill. A 200-byte /// `Hello` parked in that 1 KiB buffer never reaches the frontend. /// Flushing per-chunk turns the bridge into a true pass-through — /// each batch the kernel hands us is forwarded immediately. /// /// `Interrupted` is retried (matches `std::io::copy`'s behavior). /// Returns the same `(reader, writer)` error that `copy` would. #[cfg(test)] fn copy_with_flush(reader: &mut R, writer: &mut W) -> std::io::Result { copy_with_flush_named(reader, writer, None) } fn copy_with_flush_named( reader: &mut R, writer: &mut W, label: Option<&str>, ) -> std::io::Result { let mut buf = [0u8; 8192]; let mut total: u64 = 0; loop { match reader.read(&mut buf) { Ok(0) => return Ok(total), Ok(n) => { writer.write_all(&buf[..n])?; writer.flush()?; total += n as u64; if let Some(label) = label { bridge_debug(format!( "{label}: forwarded chunk {n} bytes (total {total})" )); } } Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} Err(e) => return Err(e), } } } /// Unbuffered writer for the bridge's protocol output fd. /// /// `std::io::Stdout` is allowed to buffer internally, which is toxic /// for `pmacs --daemon-attach`: the first daemon `Hello` is a small /// binary frame with no newline and the local frontend is blocked /// waiting for it. This writer bypasses Rust's stdout buffering and /// writes directly to fd 1 (or fd 2 for a diagnostic SSH fallback) /// using safe `nix::unistd::write`. enum RawProtocolOutput { Stdout(std::io::Stdout), Stderr(std::io::Stderr), } impl RawProtocolOutput { fn new(fd: i32) -> Self { match fd { 2 => Self::Stderr(std::io::stderr()), _ => Self::Stdout(std::io::stdout()), } } } impl Write for RawProtocolOutput { fn write(&mut self, buf: &[u8]) -> std::io::Result { match self { Self::Stdout(stdout) => { nix::unistd::write(stdout.as_fd(), buf).map_err(std::io::Error::from) } Self::Stderr(stderr) => { nix::unistd::write(stderr.as_fd(), buf).map_err(std::io::Error::from) } } } fn flush(&mut self) -> std::io::Result<()> { Ok(()) } } // Owned `PathBuf` matches the public entry-point signature; clippy's // pedantic pass-by-value is wrong for this seam. #[allow(clippy::needless_pass_by_value)] #[cfg(test)] pub(crate) fn run_bridge( socket_path: PathBuf, local_input: R, local_output: W, ) -> Result<(), BridgeError> where R: Read + Send + 'static, W: Write + Send + 'static, { let socket = UnixStream::connect(&socket_path).map_err(|source| BridgeError::Connect { socket: socket_path.clone(), source, })?; run_bridge_connected(socket, local_input, local_output) } fn run_bridge_connected( socket: UnixStream, local_input: R, mut local_output: W, ) -> Result<(), BridgeError> where R: Read + Send + 'static, W: Write + Send + 'static, { let socket_for_writer = socket.try_clone().map_err(BridgeError::Io)?; let kick_handle = socket.try_clone().map_err(BridgeError::Io)?; let mut socket_reader = socket; bridge_debug("bridge connected; starting copy loops"); let (done_tx, done_rx) = mpsc::channel::<()>(); let stdin_thread = thread::spawn(move || { let mut local_input = local_input; let mut socket_writer = socket_for_writer; bridge_debug("stdin->daemon copy loop started"); let result = copy_with_flush_named(&mut local_input, &mut socket_writer, Some("stdin->daemon")); match &result { Ok(n) => bridge_debug(format!("stdin->daemon copy loop ended after {n} bytes")), Err(e) => bridge_debug(format!("stdin->daemon copy loop failed: {e}")), } // After stdin EOF, signal write-side close so the daemon's // blocking read returns 0. Without this, the daemon never // sees EOF (we hold other socket FDs on the main thread) // and the socket→stdout copy on the main thread blocks // forever. let _ = socket_writer.shutdown(Shutdown::Write); let _ = done_tx.send(()); result }); // Main thread: forward instance bytes to local stdout. Returns // when the socket closes (EOF) or stdout breaks. Errors are // swallowed — once we've decided to tear down, the caller cares // only about whether the connect succeeded, not about the exact // shape of the disconnect. bridge_debug("daemon->stdout copy loop started"); let daemon_to_stdout = copy_with_flush_named( &mut socket_reader, &mut local_output, Some("daemon->stdout"), ); match &daemon_to_stdout { Ok(n) => bridge_debug(format!("daemon->stdout copy loop ended after {n} bytes")), Err(e) => bridge_debug(format!("daemon->stdout copy loop failed: {e}")), } let _ = daemon_to_stdout; // Wake the stdin thread's pending socket write. If the thread is // currently blocked in a *read* of local_input, this doesn't // help; we rely on the caller (or production stdin EOF) to // unblock that path. Best-effort. let _ = kick_handle.shutdown(Shutdown::Both); // Best-effort *bounded* join. Two real cases: // // * Clean detach / SSH-side EOF: the stdin thread exits within // microseconds (`copy_with_flush` returned, `shutdown(Write)` // ran, the channel send fired). The recv_timeout returns // immediately and we proceed. // * Daemon crash mid-session: the daemon is gone but the SSH // client still holds our stdin open, so the stdin thread is // wedged forever in `read(local_input)`. A pure `join()` // would block the bridge process forever, leaving the user's // `pmacs --attach` stuck because our stdout never closes. // Bound the wait so the bridge can exit; the OS reaps the // wedged thread on process termination. // // 500ms is a generous upper bound on natural exit latency // without making the crash-recovery path feel sluggish. let _ = done_rx.recv_timeout(Duration::from_millis(500)); drop(stdin_thread); bridge_debug("bridge exiting cleanly"); Ok(()) } #[cfg(test)] mod tests { use super::*; use std::os::unix::net::UnixListener; use std::sync::mpsc; use std::time::Duration; /// Try to bind a `UnixListener` at `path`. On `PermissionDenied` /// (e.g., a sandboxed CI environment that disallows `AF_UNIX` /// socket creation), prints a skip notice to stderr and returns /// `None`; the calling test should then early-return so the /// suite reports `0 failed` rather than a misleading panic. /// /// Any other `io::Error` is still a hard failure: the test /// should not silently skip on (e.g.) `EADDRINUSE` --- that's a /// real bug in the test setup. fn bind_or_skip(path: &Path) -> Option { match UnixListener::bind(path) { Ok(listener) => Some(listener), Err(e) if e.kind() == std::io::ErrorKind::PermissionDenied => { eprintln!( "test skipped: UnixListener::bind {} → PermissionDenied \ (sandboxed environment). To run this test, give the \ test process permission to create AF_UNIX sockets.", path.display() ); None } Err(e) => panic!("UnixListener::bind {} failed: {e}", path.display()), } } /// End-to-end byte echo through the bridge. /// /// Setup: /// * A `UnixListener` plays the daemon at a tempdir socket. /// * The daemon thread accepts one connection and echoes every /// byte it receives back to itself. /// * Two pipe pairs simulate the local stdin/stdout. /// * `run_bridge` runs in its own thread. /// /// Test asserts: a known sequence of bytes written to the input /// writer reappears on the output reader after the round trip /// through the daemon. #[test] fn bridge_round_trips_bytes_through_daemon() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); let Some(listener) = bind_or_skip(&socket_path) else { return; }; // Echo daemon: read everything, write it back, exit on EOF. let daemon = thread::spawn(move || { let (mut stream, _addr) = listener.accept().unwrap(); let mut stream_clone = stream.try_clone().unwrap(); let mut buf = [0u8; 1024]; loop { let n = match stream.read(&mut buf) { Ok(n) if n > 0 => n, _ => break, }; if stream_clone.write_all(&buf[..n]).is_err() { break; } } }); // Pipe pairs standing in for stdin/stdout. let (input_reader, mut input_writer) = pipe(); let (mut output_reader, output_writer) = pipe(); let bridge_socket = socket_path.clone(); let bridge = thread::spawn(move || run_bridge(bridge_socket, input_reader, output_writer)); // Round-trip a known sequence. let payload = b"hello, bridge\n"; input_writer.write_all(payload).unwrap(); input_writer.flush().unwrap(); let mut received = vec![0u8; payload.len()]; // Read exactly payload.len() bytes back. read_exact retries // on Interrupted but will return UnexpectedEof if the bridge // closes early — that would fail the test loudly. output_reader.read_exact(&mut received).unwrap(); assert_eq!(&received, payload); // Tear down: drop the input writer to EOF the stdin thread, // which closes the socket-side write. The daemon notices, // closes its end, the socket-side EOF reaches the bridge's // socket_reader, the main `copy` returns, run_bridge returns. drop(input_writer); // Bound the wait for run_bridge to return; if anything // deadlocks we want the test to fail with a clear message // rather than hang the whole suite. let (done_tx, done_rx) = mpsc::channel(); thread::spawn(move || { let result = bridge.join(); let _ = done_tx.send(result); }); let bridge_result = done_rx .recv_timeout(Duration::from_secs(2)) .expect("run_bridge should return within 2s after EOF"); let bridge_result = bridge_result.expect("bridge thread did not panic"); assert!( bridge_result.is_ok(), "run_bridge should return Ok on clean EOF, got {bridge_result:?}", ); let _ = daemon.join(); } #[test] fn connect_failure_yields_workaround_pointing_message() { // No daemon running here. The Connect error message tells the // user how to recover. let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("nonexistent.sock"); let (input_reader, _input_writer) = pipe(); let (_output_reader, output_writer) = pipe(); let result = run_bridge(socket_path.clone(), input_reader, output_writer); match result { Err(BridgeError::Connect { socket, source: _ }) => { assert_eq!(socket, socket_path); let msg = format!( "{}", BridgeError::Connect { socket: socket.clone(), source: std::io::Error::other("test"), } ); assert!( msg.contains("pmacs --daemon"), "error message must point at the workaround: {msg}", ); } other => panic!("expected Connect error, got {other:?}"), } } /// In-process pipe pair using a `UnixStream::pair`. Each half is /// `Read + Write + Send + 'static`; tests only use one direction /// of each. fn pipe() -> (UnixStream, UnixStream) { UnixStream::pair().expect("UnixStream::pair") } /// Writer that buffers internally and only "delivers" on explicit /// flush — models `std::io::Stdout`'s `LineWriter` against binary /// payloads (no newlines means `write` alone never reaches the /// reader). Surfaces buffered-but-unflushed bytes as a discrepancy /// between `delivered.len()` and the post-write byte count. struct FlushOnlyWriter { buffer: Vec, delivered: Arc>>, } impl Write for FlushOnlyWriter { fn write(&mut self, buf: &[u8]) -> std::io::Result { self.buffer.extend_from_slice(buf); Ok(buf.len()) } fn flush(&mut self) -> std::io::Result<()> { self.delivered .lock() .unwrap() .extend_from_slice(&self.buffer); self.buffer.clear(); Ok(()) } } /// Regression test for the `LineWriter` / buffered-stdout bug: /// `copy_with_flush` must call `flush` after every chunk, not /// just rely on `write_all`. Without the per-chunk flush, the /// production `pmacs --daemon-attach` parked a 200-byte `Hello` /// frame in stdout's `LineWriter` buffer and the frontend never /// saw it until the buffer filled. #[test] fn copy_with_flush_delivers_each_chunk_without_waiting_for_buffer_fill() { let payload: &[u8] = &[0xAA; 200]; // postcard-shaped binary, no '\n' let mut reader: &[u8] = payload; let delivered = Arc::new(Mutex::new(Vec::new())); let mut writer = FlushOnlyWriter { buffer: Vec::new(), delivered: delivered.clone(), }; let total = copy_with_flush(&mut reader, &mut writer).expect("copy ok"); assert_eq!(total, payload.len() as u64); assert_eq!( delivered.lock().unwrap().as_slice(), payload, "every chunk must be flushed (else the bridge stalls on small binary frames)", ); assert!( writer.buffer.is_empty(), "no bytes left unflushed in the writer's internal buffer", ); } // ----------------------------------------------------------------- // M5.7c — auto-start helper tests // ----------------------------------------------------------------- use std::sync::Arc; use std::sync::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; #[test] fn ensure_running_returns_immediately_when_daemon_already_listening() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); let Some(_listener) = bind_or_skip(&socket_path) else { return; }; let spawner_called = Arc::new(AtomicBool::new(false)); let sc = spawner_called.clone(); let result = ensure_daemon_running_with( &socket_path, Duration::from_secs(1), move |_path: &Path| -> Result<(), BridgeError> { sc.store(true, Ordering::SeqCst); Ok(()) }, ); assert!(result.is_ok(), "fast path should return Ok, got {result:?}"); assert!( !spawner_called.load(Ordering::SeqCst), "spawner must not be invoked when daemon is already listening", ); } #[test] fn ensure_running_reuses_existing_connection_instead_of_probe_drop() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); let Some(listener) = bind_or_skip(&socket_path) else { return; }; let daemon = thread::spawn(move || { let (mut stream, _addr) = listener.accept().unwrap(); stream.write_all(b"HELLO").unwrap(); stream.flush().unwrap(); }); let mut stream = ensure_daemon_running_with( &socket_path, Duration::from_secs(1), |_path: &Path| -> Result<(), BridgeError> { panic!("spawner must not run when daemon is already listening") }, ) .expect("existing daemon connection"); let mut received = [0u8; 5]; stream.read_exact(&mut received).unwrap(); assert_eq!(&received, b"HELLO"); daemon.join().unwrap(); } #[test] fn ensure_running_invokes_spawner_then_waits_for_socket_to_appear() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); // Probe-bind once up front. The actual bind happens inside a // child thread (and we cannot return from the test from // there), so checking PermissionDenied here lets us skip // before scheduling the worker. match bind_or_skip(&socket_path) { Some(listener) => drop(listener), None => return, } // Some kernels keep the inode visible after drop. Make sure // the path is gone so the spawner's bind doesn't EADDRINUSE. let _ = std::fs::remove_file(&socket_path); // Spawner: defer binding by 100ms in a worker thread, then // hold the listener long enough for `ensure_running` to see // it. The spawner returns Ok as soon as the worker is // launched; the polling in `ensure_running` does the rest. let spawner = move |path: &Path| -> Result<(), BridgeError> { let path = path.to_path_buf(); thread::spawn(move || { thread::sleep(Duration::from_millis(100)); let listener = UnixListener::bind(&path).unwrap(); // Hold the listener until the test should be done. thread::sleep(Duration::from_millis(500)); drop(listener); }); Ok(()) }; let start = Instant::now(); let result = ensure_daemon_running_with(&socket_path, Duration::from_secs(2), spawner); let elapsed = start.elapsed(); assert!(result.is_ok(), "expected Ok, got {result:?}"); assert!( elapsed >= Duration::from_millis(100), "should have waited ~100ms for the worker to bind, took {elapsed:?}", ); assert!( elapsed < Duration::from_secs(2), "should not have hit the timeout; took {elapsed:?}", ); } #[test] fn ensure_running_times_out_when_socket_never_binds() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); // No-op spawner: returns Ok but never binds the socket. // ensure_running should poll until the timeout fires. let spawner = |_path: &Path| -> Result<(), BridgeError> { Ok(()) }; let timeout = Duration::from_millis(200); let start = Instant::now(); let result = ensure_daemon_running_with(&socket_path, timeout, spawner); let elapsed = start.elapsed(); match result { Err(BridgeError::AutoStartTimeout { socket, after }) => { assert_eq!(socket, socket_path); assert_eq!(after, timeout); } other => panic!("expected AutoStartTimeout, got {other:?}"), } assert!( elapsed >= timeout, "should have waited at least {timeout:?}, took {elapsed:?}", ); // Loose upper bound: timeout + one poll interval + scheduling // slack. assert!( elapsed < timeout + Duration::from_millis(500), "should have returned shortly after the timeout, took {elapsed:?}", ); } #[test] fn ensure_running_propagates_spawner_error() { let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("test.sock"); // Spawner that fails outright. The error must bubble up // unchanged so users see *why* spawn failed (not just "your // daemon never showed up"). let spawner = |_path: &Path| -> Result<(), BridgeError> { Err(BridgeError::SpawnFailed { exe: PathBuf::from("/path/to/test"), source: std::io::Error::other("test forced spawn failure"), }) }; let result = ensure_daemon_running_with(&socket_path, Duration::from_secs(1), spawner); match result { Err(BridgeError::SpawnFailed { exe, source }) => { assert_eq!(exe, PathBuf::from("/path/to/test")); assert!(format!("{source}").contains("test forced spawn failure")); } other => panic!("expected SpawnFailed, got {other:?}"), } } #[test] fn auto_start_timeout_message_points_at_manual_recovery() { let err = BridgeError::AutoStartTimeout { socket: PathBuf::from("/tmp/x.sock"), after: Duration::from_secs(5), }; let msg = format!("{err}"); assert!( msg.contains("pmacs --daemon"), "timeout message must point at manual recovery: {msg}", ); assert!(msg.contains("/tmp/x.sock")); assert!(msg.contains("5.0s")); } #[test] fn spawn_failed_message_includes_exe_path_and_workaround_pointer() { let err = BridgeError::SpawnFailed { exe: PathBuf::from("/usr/local/bin/pmacs"), source: std::io::Error::other("permission denied"), }; let msg = format!("{err}"); assert!(msg.contains("/usr/local/bin/pmacs")); assert!(msg.contains("permission denied")); assert!(msg.contains("PATH"), "should hint at PATH check: {msg}"); } /// F8 reproduction (M10.11 acceptance criterion 1, SSH transport). /// /// Models the production failure: the daemon speaks first /// (`per_attach_thread` writes `Hello` *before* reading /// `AttachRequest`, see `src/daemon.rs:617`), and the bridge's /// local input reaches EOF immediately — exactly what /// `std::io::empty()` provides, and what was observed when /// `pmacs --attach ssh:...` drove the remote `pmacs /// --daemon-attach`. /// /// The daemon thread deliberately sleeps before writing `Hello` /// so the bridge's immediate-stdin-EOF teardown path /// (`copy_with_flush` returns `Ok(0)` → `shutdown(Write)` → /// `done_tx`) wins the race, reproducing the production ordering /// where the daemon's first write lands *after* the bridge has /// reacted to EOF. /// /// Acceptance: the daemon's `Hello` write must NOT fail with /// `BrokenPipe`, and the `Hello` bytes must reach the bridge's /// local output. If this test fails, F8 is reproduced in-process /// and the defect is in `run_bridge`'s local teardown logic — not /// ssh-specific. If it passes against unfixed code, the in-process /// model does NOT capture F8 and the cause is ssh-transport /// specific (report honestly; do not claim a `run_bridge` fix). #[test] fn f8_daemon_speaks_first_survives_immediate_stdin_eof() { // Length-prefixed Hello-shaped frame. Content is irrelevant — // the bridge is byte-transparent; only survival matters. const HELLO_FRAME: &[u8] = &[0x00, 0x00, 0x00, 0x05, b'H', b'E', b'L', b'L', b'O']; let tmp = tempfile::tempdir().unwrap(); let socket_path = tmp.path().join("f8.sock"); let Some(listener) = bind_or_skip(&socket_path) else { return; }; let (write_result_tx, write_result_rx) = mpsc::channel::>(); let daemon = thread::spawn(move || { let (mut stream, _addr) = listener.accept().unwrap(); // Production ordering: the bridge has process/network // latency before our first write reaches it. Sleep so the // bridge's immediate-EOF teardown definitely happens first. thread::sleep(Duration::from_millis(150)); let result = stream.write_all(HELLO_FRAME).and_then(|()| stream.flush()); let _ = write_result_tx.send(result); // Mirror per_attach_thread: after Hello, read AttachRequest // (here just drain to EOF so the socket lifecycle matches). let mut sink = Vec::new(); let _ = stream.read_to_end(&mut sink); }); let (mut output_reader, output_writer) = pipe(); let bridge_socket = socket_path.clone(); // std::io::empty() == immediate stdin EOF. let bridge = thread::spawn(move || run_bridge(bridge_socket, std::io::empty(), output_writer)); // 1. The daemon's Hello write must not EPIPE. let write_result = write_result_rx .recv_timeout(Duration::from_secs(3)) .expect("daemon thread should attempt the Hello write within 3s"); assert!( write_result.is_ok(), "F8 reproduced: daemon's Hello write failed ({:?}) — \ run_bridge tore the socket down on immediate stdin-EOF \ before the daemon's first write could land", write_result.as_ref().err().map(std::io::Error::kind), ); // 2. The Hello bytes must reach the bridge's local output. // macOS rejects read timeouts on UnixStream::pair with EINVAL, // so use a reader thread plus channel timeout for portability. let (read_tx, read_rx) = mpsc::channel(); thread::spawn(move || { let mut received = vec![0u8; HELLO_FRAME.len()]; let result = output_reader.read_exact(&mut received).map(|()| received); let _ = read_tx.send(result); }); match read_rx .recv_timeout(Duration::from_secs(3)) .expect("reader thread should return within 3s") { Ok(received) => assert_eq!(received, HELLO_FRAME, "Hello bytes corrupted in transit"), Err(e) => panic!( "F8 reproduced: Hello never reached local output ({e}) — \ the bridge dropped the daemon→client direction on \ immediate stdin-EOF", ), } // Bounded teardown so a deadlock fails loudly, not silently. let (done_tx, done_rx) = mpsc::channel(); thread::spawn(move || { let _ = done_tx.send(bridge.join()); }); let _ = done_rx.recv_timeout(Duration::from_secs(3)); let _ = daemon.join(); } }