merge: tty7_core::client — public control and pane clients

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014JPaaZVK7rfQPKyrymzsYv

crates/tty7-core/src/client/control.rs

@@ -0,0 +1,405 @@
+use std::io;
+use std::path::Path;
+use std::sync::Mutex;
+use std::sync::mpsc::{Receiver, channel};
+use std::time::Duration;
+
+use crate::daemon::control::{
+    ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ControlResponse, EventSink, ReplyOk,
+};
+use crate::daemon::router::{RouteAction, RouteChannel, RouteHeader, RouteTarget, negotiate};
+use crate::daemon::transport;
+
+pub struct ControlClient {
+    link: crate::daemon::control::ControlClient,
+    events: Mutex<Receiver<ControlEvent>>,
+}
+
+impl ControlClient {
+    pub fn connect(hello: &ControlHello) -> io::Result<ControlClient> {
+        Self::over_stream(connect_local_control()?, hello)
+    }
+
+    pub fn connect_at(endpoint: &Path, hello: &ControlHello) -> io::Result<ControlClient> {
+        Self::over_stream(transport::connect_endpoint_at(endpoint)?, hello)
+    }
+
+    pub fn routed(target: RouteTarget, hello: &ControlHello) -> io::Result<ControlClient> {
+        Self::routed_over(transport::connect()?, target, hello)
+    }
+
+    pub fn routed_over(
+        mut stream: transport::Stream,
+        target: RouteTarget,
+        hello: &ControlHello,
+    ) -> io::Result<ControlClient> {
+        let header = RouteHeader {
+            target,
+            server_command: None,
+            channel: RouteChannel::Control,
+            action: RouteAction::Forward,
+        };
+        negotiate(&mut stream, &header)?;
+        Self::over_stream(stream, hello)
+    }
+
+    pub fn over_stream(
+        stream: transport::Stream,
+        hello: &ControlHello,
+    ) -> io::Result<ControlClient> {
+        let (push, events) = channel();
+        let sink: EventSink = Box::new(move |event| {
+            let _ = push.send(event);
+        });
+        #[cfg(unix)]
+        let link = crate::daemon::control::ControlClient::over_unix(stream, hello, sink)?;
+        #[cfg(windows)]
+        let link = crate::daemon::control::ControlClient::over_tcp(stream, hello, sink)?;
+        Ok(ControlClient {
+            link,
+            events: Mutex::new(events),
+        })
+    }
+
+    pub fn hello(&self) -> &ControlHelloOk {
+        self.link.hello()
+    }
+
+    pub fn is_connected(&self) -> bool {
+        self.link.is_connected()
+    }
+
+    pub fn request(&self, req: ControlRequest) -> io::Result<ReplyOk> {
+        self.link.call(req)
+    }
+
+    pub fn request_full(&self, req: ControlRequest, blob: &[u8]) -> io::Result<ControlResponse> {
+        self.link.call_full(req, blob)
+    }
+
+    pub fn request_with_deadline(
+        &self,
+        req: ControlRequest,
+        blob: &[u8],
+        deadline: Duration,
+    ) -> io::Result<ControlResponse> {
+        self.link.call_with_deadline(req, blob, deadline)
+    }
+
+    pub fn next_event(&self, wait: Duration) -> Option<ControlEvent> {
+        let events = self.events.lock().ok()?;
+        events.recv_timeout(wait).ok()
+    }
+
+    pub fn events(&self) -> ControlEvents<'_> {
+        ControlEvents { client: self }
+    }
+
+    pub fn close(&self) {
+        self.link.close();
+    }
+}
+
+impl std::fmt::Debug for ControlClient {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.link.fmt(f)
+    }
+}
+
+pub struct ControlEvents<'a> {
+    client: &'a ControlClient,
+}
+
+impl Iterator for ControlEvents<'_> {
+    type Item = ControlEvent;
+
+    fn next(&mut self) -> Option<ControlEvent> {
+        let events = self.client.events.lock().ok()?;
+        events.recv().ok()
+    }
+}
+
+fn connect_local_control() -> io::Result<transport::Stream> {
+    #[cfg(unix)]
+    {
+        let path = crate::host::server::control_socket_path()?;
+        transport::connect_endpoint_at(&path)
+    }
+    #[cfg(windows)]
+    {
+        crate::host::server::connect_control()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::client::stream_pair;
+    use crate::daemon::control::{
+        CONTROL_VERSION, ControlClientMsg, ControlReply, ControlServerMsg, feature,
+    };
+    use crate::daemon::protocol::{read_frame, write_frame};
+    use crate::daemon::router::{ROUTE_KIND, RouteAck};
+    use crate::host::{MTime, Meta};
+    use std::io::{Read, Write};
+
+    const EVENT_WAIT: Duration = Duration::from_secs(10);
+
+    fn hello() -> ControlHello {
+        ControlHello::host_rpc("unit-token", "unit-host")
+    }
+
+    fn hello_ok() -> crate::daemon::control::ControlHelloOk {
+        crate::daemon::control::ControlHelloOk {
+            control_version: CONTROL_VERSION,
+            protocol_version: crate::daemon::protocol::PROTOCOL_VERSION,
+            build: "unit".into(),
+            separator: '/',
+            home: "/home/unit".into(),
+            features: vec![feature::CONTROL.into()],
+            instance: "unit-instance".into(),
+        }
+    }
+
+    fn meta() -> Meta {
+        Meta {
+            is_dir: false,
+            is_symlink: false,
+            len: 8,
+            mtime: Some(MTime {
+                secs: 1_769_000_000,
+                nanos: 0,
+            }),
+            readonly: false,
+        }
+    }
+
+    fn answer_hello(r: &mut impl Read, w: &mut impl Write) {
+        match ControlClientMsg::read(r).expect("read the client hello") {
+            ControlClientMsg::Hello(_) => {}
+            other => panic!("expected HELLO first, got {other:?}"),
+        }
+        ControlServerMsg::HelloOk(hello_ok())
+            .encode(w)
+            .expect("answer HELLO_OK");
+    }
+
+    #[test]
+    fn requests_events_and_blobs_flow_through_the_wrapper() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            answer_hello(&mut r, &mut w);
+
+            match ControlClientMsg::read(&mut r).expect("read Ping") {
+                ControlClientMsg::Request {
+                    req_id,
+                    req: ControlRequest::Ping,
+                } => {
+                    ControlServerMsg::Event(ControlEvent::LayoutResync)
+                        .encode(&mut w)
+                        .expect("push an event");
+                    ControlServerMsg::Response {
+                        req_id,
+                        reply: ControlReply::Ok(ReplyOk::Pong),
+                    }
+                    .encode(&mut w)
+                    .expect("answer Ping");
+                }
+                other => panic!("expected Ping, got {other:?}"),
+            }
+
+            match ControlClientMsg::read(&mut r).expect("read WriteFile") {
+                ControlClientMsg::RequestBlob {
+                    req_id,
+                    req: ControlRequest::WriteFile { .. },
+                    blob,
+                } => {
+                    assert_eq!(blob, b"payload", "the request blob rides the frame");
+                    ControlServerMsg::Response {
+                        req_id,
+                        reply: ControlReply::Ok(ReplyOk::Unit),
+                    }
+                    .encode(&mut w)
+                    .expect("answer WriteFile");
+                }
+                other => panic!("expected a WriteFile blob, got {other:?}"),
+            }
+
+            match ControlClientMsg::read(&mut r).expect("read ReadFile") {
+                ControlClientMsg::Request {
+                    req_id,
+                    req: ControlRequest::ReadFile { .. },
+                } => {
+                    ControlServerMsg::ResponseBlob {
+                        req_id,
+                        reply: ControlReply::Ok(ReplyOk::FileMeta { meta: meta() }),
+                        blob: b"contents".to_vec(),
+                    }
+                    .encode(&mut w)
+                    .expect("answer ReadFile");
+                }
+                other => panic!("expected ReadFile, got {other:?}"),
+            }
+        });
+
+        let client = ControlClient::over_stream(client_end, &hello()).expect("handshake");
+        assert!(client.hello().has_feature(feature::CONTROL));
+
+        assert!(matches!(
+            client.request(ControlRequest::Ping).expect("ping"),
+            ReplyOk::Pong
+        ));
+        assert!(matches!(
+            client.next_event(EVENT_WAIT),
+            Some(ControlEvent::LayoutResync)
+        ));
+
+        let wrote = client
+            .request_full(
+                ControlRequest::WriteFile {
+                    path: "/tmp/x".into(),
+                },
+                b"payload",
+            )
+            .expect("write");
+        assert!(matches!(wrote.reply, ReplyOk::Unit));
+
+        let read = client
+            .request_full(
+                ControlRequest::ReadFile {
+                    path: "/tmp/x".into(),
+                    max_bytes: 1024,
+                },
+                &[],
+            )
+            .expect("read");
+        assert!(matches!(read.reply, ReplyOk::FileMeta { .. }));
+        assert_eq!(read.blob, b"contents", "the reply blob comes back intact");
+
+        client.close();
+        server.join().expect("server thread");
+    }
+
+    #[test]
+    fn the_event_iterator_ends_when_the_link_goes_down() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            answer_hello(&mut r, &mut w);
+            ControlServerMsg::Event(ControlEvent::PaneExited {
+                pane_id: 3,
+                code: Some(0),
+            })
+            .encode(&mut w)
+            .expect("push an event");
+        });
+
+        let client = ControlClient::over_stream(client_end, &hello()).expect("handshake");
+        server.join().expect("server thread");
+
+        let mut events = client.events();
+        assert!(matches!(
+            events.next(),
+            Some(ControlEvent::PaneExited {
+                pane_id: 3,
+                code: Some(0)
+            })
+        ));
+        assert!(
+            events.next().is_none(),
+            "a dead link must end the iterator, not block it"
+        );
+    }
+
+    #[test]
+    fn routed_over_negotiates_before_the_handshake() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+
+            let (kind, payload) = read_frame(&mut r).expect("read the route header");
+            assert_eq!(kind, ROUTE_KIND, "the ROUTE frame must come first");
+            let header = RouteHeader::decode(&payload).expect("decode the header");
+            assert_eq!(header.channel, RouteChannel::Control);
+            assert!(matches!(header.target, RouteTarget::Wsl { ref distro } if distro == "Ubuntu"));
+
+            let ack = serde_json::to_vec(&RouteAck {
+                ok: true,
+                link: Some("unit".into()),
+                action: Some(RouteAction::Forward),
+                error: None,
+            })
+            .expect("encode the ack");
+            write_frame(&mut w, ROUTE_KIND, &ack).expect("send the ack");
+
+            answer_hello(&mut r, &mut w);
+            match ControlClientMsg::read(&mut r).expect("read Ping") {
+                ControlClientMsg::Request {
+                    req_id,
+                    req: ControlRequest::Ping,
+                } => {
+                    ControlServerMsg::Response {
+                        req_id,
+                        reply: ControlReply::Ok(ReplyOk::Pong),
+                    }
+                    .encode(&mut w)
+                    .expect("answer Ping");
+                }
+                other => panic!("expected Ping, got {other:?}"),
+            }
+        });
+
+        let client = ControlClient::routed_over(
+            client_end,
+            RouteTarget::Wsl {
+                distro: "Ubuntu".into(),
+            },
+            &hello(),
+        )
+        .expect("routed handshake");
+        assert!(matches!(
+            client.request(ControlRequest::Ping).expect("ping"),
+            ReplyOk::Pong
+        ));
+
+        client.close();
+        server.join().expect("server thread");
+    }
+
+    #[test]
+    fn a_refused_route_surfaces_the_servers_reason() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            let (kind, _) = read_frame(&mut r).expect("read the route header");
+            assert_eq!(kind, ROUTE_KIND);
+            let ack = serde_json::to_vec(&RouteAck {
+                ok: false,
+                link: None,
+                action: None,
+                error: Some("no such distro".into()),
+            })
+            .expect("encode the refusal");
+            write_frame(&mut w, ROUTE_KIND, &ack).expect("send the refusal");
+        });
+
+        let err = ControlClient::routed_over(
+            client_end,
+            RouteTarget::Wsl {
+                distro: "Nowhere".into(),
+            },
+            &hello(),
+        )
+        .expect_err("a refused route must not hand back a client");
+        assert!(
+            err.to_string().contains("no such distro"),
+            "the refusal reason was lost: {err}"
+        );
+        server.join().expect("server thread");
+    }
+}

crates/tty7-core/src/client/mod.rs

@@ -0,0 +1,31 @@
+mod control;
+mod pane;
+
+pub use control::{ControlClient, ControlEvents};
+pub use pane::{PaneClient, PaneInput, PaneOutput, PaneSession};
+
+pub use crate::daemon::control::{
+    ControlEvent, ControlHello, ControlHelloOk, ControlRequest, ControlResponse, ReplyOk,
+};
+pub use crate::daemon::protocol::{DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize};
+pub use crate::daemon::router::RouteTarget;
+
+#[cfg(test)]
+pub(crate) fn stream_pair() -> (
+    crate::daemon::transport::Stream,
+    crate::daemon::transport::Stream,
+) {
+    #[cfg(unix)]
+    {
+        std::os::unix::net::UnixStream::pair().expect("socketpair")
+    }
+    #[cfg(windows)]
+    {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback");
+        let addr = listener.local_addr().expect("bound addr");
+        let connecting =
+            std::thread::spawn(move || std::net::TcpStream::connect(addr).expect("connect back"));
+        let (accepted, _) = listener.accept().expect("accept");
+        (connecting.join().expect("connector thread"), accepted)
+    }
+}

crates/tty7-core/src/client/pane.rs

@@ -0,0 +1,434 @@
+use std::io::{self, Read as _};
+use std::path::PathBuf;
+use std::time::Duration;
+
+use crate::daemon::protocol::{
+    ClientMsg, DaemonMsg, DaemonVersion, PaneInfo, ShellSpec, WinSize, is_error_kind,
+    peek_frame_kind, take_frame,
+};
+use crate::daemon::transport;
+
+const OPEN_REPLY_WAIT: Duration = Duration::from_secs(15);
+
+#[derive(Clone, Debug, Default)]
+enum PaneEndpoint {
+    #[default]
+    Local,
+    At(PathBuf),
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct PaneClient {
+    endpoint: PaneEndpoint,
+}
+
+impl PaneClient {
+    pub fn local() -> PaneClient {
+        PaneClient {
+            endpoint: PaneEndpoint::Local,
+        }
+    }
+
+    pub fn at(endpoint: impl Into<PathBuf>) -> PaneClient {
+        PaneClient {
+            endpoint: PaneEndpoint::At(endpoint.into()),
+        }
+    }
+
+    fn open(&self) -> io::Result<transport::Stream> {
+        match &self.endpoint {
+            PaneEndpoint::Local => transport::connect(),
+            PaneEndpoint::At(path) => transport::connect_endpoint_at(path),
+        }
+    }
+
+    pub fn list(&self) -> io::Result<Vec<PaneInfo>> {
+        let mut stream = self.open()?;
+        ClientMsg::List.encode(&mut stream)?;
+        match DaemonMsg::read(&mut stream)? {
+            DaemonMsg::PaneList(panes) => Ok(panes),
+            DaemonMsg::Error(message) => Err(io::Error::other(message)),
+            other => Err(unexpected_reply("List", &other)),
+        }
+    }
+
+    pub fn version(&self) -> io::Result<DaemonVersion> {
+        let mut stream = self.open()?;
+        ClientMsg::Version.encode(&mut stream)?;
+        match DaemonMsg::read(&mut stream)? {
+            DaemonMsg::Version(version) => Ok(version),
+            DaemonMsg::Error(message) => Err(io::Error::other(message)),
+            other => Err(unexpected_reply("Version", &other)),
+        }
+    }
+
+    pub fn kill(&self, pane_id: u64) -> io::Result<()> {
+        let mut stream = self.open()?;
+        ClientMsg::Kill { pane_id }.encode(&mut stream)
+    }
+
+    pub fn spawn(
+        &self,
+        cwd: Option<PathBuf>,
+        size: WinSize,
+        shell: Option<ShellSpec>,
+        owner: Option<String>,
+    ) -> io::Result<PaneSession> {
+        PaneSession::spawn_over(self.open()?, cwd, size, shell, owner, OPEN_REPLY_WAIT)
+    }
+
+    pub fn attach(&self, pane_id: u64, size: WinSize) -> io::Result<PaneSession> {
+        PaneSession::attach_over(self.open()?, pane_id, size, OPEN_REPLY_WAIT)
+    }
+}
+
+#[derive(Debug)]
+pub struct PaneSession {
+    input: PaneInput,
+    output: PaneOutput,
+}
+
+impl PaneSession {
+    pub(crate) fn spawn_over(
+        mut stream: transport::Stream,
+        cwd: Option<PathBuf>,
+        size: WinSize,
+        shell: Option<ShellSpec>,
+        owner: Option<String>,
+        reply_wait: Duration,
+    ) -> io::Result<PaneSession> {
+        ClientMsg::Spawn {
+            cwd,
+            size,
+            shell,
+            owner,
+        }
+        .encode(&mut stream)?;
+        let mut session = PaneSession::over(stream, 0)?;
+        session.set_recv_timeout(Some(reply_wait))?;
+        let first = session.recv();
+        session.set_recv_timeout(None)?;
+        match first {
+            Ok(DaemonMsg::Spawned { pane_id }) => {
+                session.input.pane_id = pane_id;
+                Ok(session)
+            }
+            Ok(DaemonMsg::Error(message)) => {
+                Err(io::Error::other(format!("daemon refused Spawn: {message}")))
+            }
+            Ok(other) => Err(unexpected_reply("Spawn", &other)),
+            Err(e) if would_block(&e) => Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                format!("no answer to Spawn within {reply_wait:?}"),
+            )),
+            Err(e) => Err(e),
+        }
+    }
+
+    pub(crate) fn attach_over(
+        mut stream: transport::Stream,
+        pane_id: u64,
+        size: WinSize,
+        reply_wait: Duration,
+    ) -> io::Result<PaneSession> {
+        ClientMsg::Attach { pane_id, size }.encode(&mut stream)?;
+        let mut session = PaneSession::over(stream, pane_id)?;
+        session.set_recv_timeout(Some(reply_wait))?;
+        let verdict = session.output.refusal_check(pane_id);
+        session.set_recv_timeout(None)?;
+        verdict?;
+        Ok(session)
+    }
+
+    fn over(stream: transport::Stream, pane_id: u64) -> io::Result<PaneSession> {
+        let reader = stream.try_clone()?;
+        Ok(PaneSession {
+            input: PaneInput {
+                writer: stream,
+                pane_id,
+            },
+            output: PaneOutput {
+                reader,
+                buffered: Vec::new(),
+            },
+        })
+    }
+
+    pub fn pane_id(&self) -> u64 {
+        self.input.pane_id()
+    }
+
+    pub fn input(&mut self, bytes: &[u8]) -> io::Result<()> {
+        self.input.input(bytes)
+    }
+
+    pub fn resize(&mut self, size: WinSize) -> io::Result<()> {
+        self.input.resize(size)
+    }
+
+    pub fn detach(self) -> io::Result<()> {
+        self.input.detach()
+    }
+
+    pub fn kill(self) -> io::Result<()> {
+        self.input.kill()
+    }
+
+    pub fn recv(&mut self) -> io::Result<DaemonMsg> {
+        self.output.recv()
+    }
+
+    pub fn set_recv_timeout(&self, wait: Option<Duration>) -> io::Result<()> {
+        self.output.set_recv_timeout(wait)
+    }
+
+    pub fn split(self) -> (PaneInput, PaneOutput) {
+        (self.input, self.output)
+    }
+}
+
+#[derive(Debug)]
+pub struct PaneInput {
+    writer: transport::Stream,
+    pane_id: u64,
+}
+
+impl PaneInput {
+    pub fn pane_id(&self) -> u64 {
+        self.pane_id
+    }
+
+    pub fn input(&mut self, bytes: &[u8]) -> io::Result<()> {
+        ClientMsg::Input(bytes.to_vec()).encode(&mut self.writer)
+    }
+
+    pub fn resize(&mut self, size: WinSize) -> io::Result<()> {
+        ClientMsg::Resize(size).encode(&mut self.writer)
+    }
+
+    pub fn detach(mut self) -> io::Result<()> {
+        ClientMsg::Detach.encode(&mut self.writer)
+    }
+
+    pub fn kill(mut self) -> io::Result<()> {
+        ClientMsg::Kill {
+            pane_id: self.pane_id,
+        }
+        .encode(&mut self.writer)
+    }
+}
+
+#[derive(Debug)]
+pub struct PaneOutput {
+    reader: transport::Stream,
+    buffered: Vec<u8>,
+}
+
+impl PaneOutput {
+    pub fn recv(&mut self) -> io::Result<DaemonMsg> {
+        loop {
+            if let Some((kind, payload)) = take_frame(&mut self.buffered)? {
+                return DaemonMsg::from_frame(kind, payload);
+            }
+            let mut scratch = [0u8; 16 * 1024];
+            match self.reader.read(&mut scratch) {
+                Ok(0) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::UnexpectedEof,
+                        "the daemon closed the pane connection",
+                    ));
+                }
+                Ok(n) => self.buffered.extend_from_slice(&scratch[..n]),
+                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
+    pub fn set_recv_timeout(&self, wait: Option<Duration>) -> io::Result<()> {
+        self.reader.set_read_timeout(wait)
+    }
+
+    fn refusal_check(&mut self, pane_id: u64) -> io::Result<()> {
+        let mut scratch = [0u8; 4096];
+        loop {
+            if let Some(kind) = peek_frame_kind(&self.buffered) {
+                if !is_error_kind(kind) {
+                    return Ok(());
+                }
+                return Err(self.refusal(pane_id));
+            }
+            match self.reader.read(&mut scratch) {
+                Ok(0) => {
+                    return Err(io::Error::new(
+                        io::ErrorKind::UnexpectedEof,
+                        format!(
+                            "the daemon closed the connection without answering \
+                             Attach for pane {pane_id}"
+                        ),
+                    ));
+                }
+                Ok(n) => self.buffered.extend_from_slice(&scratch[..n]),
+                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                Err(e) if would_block(&e) => return Ok(()),
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
+    fn refusal(&mut self, pane_id: u64) -> io::Error {
+        match self.recv() {
+            Ok(DaemonMsg::Error(message)) => {
+                io::Error::other(format!("daemon refused Attach: {message}"))
+            }
+            Ok(other) => unexpected_reply("Attach", &other),
+            Err(_) => io::Error::other(format!("daemon refused Attach for pane {pane_id}")),
+        }
+    }
+}
+
+fn would_block(e: &io::Error) -> bool {
+    matches!(
+        e.kind(),
+        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+    )
+}
+
+fn unexpected_reply(request: &str, got: &DaemonMsg) -> io::Error {
+    io::Error::new(
+        io::ErrorKind::InvalidData,
+        format!("unexpected daemon reply to {request}: {got:?}"),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::client::stream_pair;
+
+    const REPLY_WAIT: Duration = Duration::from_secs(10);
+
+    fn size() -> WinSize {
+        WinSize {
+            cols: 80,
+            rows: 24,
+            cell_w: 8,
+            cell_h: 17,
+        }
+    }
+
+    #[test]
+    fn a_session_speaks_the_pane_protocol_end_to_end() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+
+            match ClientMsg::read(&mut r).expect("read Spawn") {
+                ClientMsg::Spawn {
+                    owner: Some(owner), ..
+                } => assert_eq!(owner, "unit"),
+                other => panic!("expected an owned Spawn, got {other:?}"),
+            }
+            DaemonMsg::Spawned { pane_id: 7 }
+                .encode(&mut w)
+                .expect("confirm the spawn");
+            DaemonMsg::Output(b"hello from the pane".to_vec())
+                .encode(&mut w)
+                .expect("stream output");
+
+            match ClientMsg::read(&mut r).expect("read Input") {
+                ClientMsg::Input(bytes) => assert_eq!(bytes, b"ls\r"),
+                other => panic!("expected Input, got {other:?}"),
+            }
+            match ClientMsg::read(&mut r).expect("read Resize") {
+                ClientMsg::Resize(new_size) => assert_eq!(new_size.cols, 120),
+                other => panic!("expected Resize, got {other:?}"),
+            }
+            DaemonMsg::Exited { code: Some(0) }
+                .encode(&mut w)
+                .expect("report the exit");
+            match ClientMsg::read(&mut r).expect("read Detach") {
+                ClientMsg::Detach => {}
+                other => panic!("expected Detach, got {other:?}"),
+            }
+        });
+
+        let mut session =
+            PaneSession::spawn_over(client_end, None, size(), None, Some("unit".into()), REPLY_WAIT)
+                .expect("spawn");
+        assert_eq!(session.pane_id(), 7);
+
+        match session.recv().expect("first stream message") {
+            DaemonMsg::Output(bytes) => assert_eq!(bytes, b"hello from the pane"),
+            other => panic!("expected Output, got {other:?}"),
+        }
+
+        session.input(b"ls\r").expect("send input");
+        session
+            .resize(WinSize {
+                cols: 120,
+                ..size()
+            })
+            .expect("send resize");
+        match session.recv().expect("exit message") {
+            DaemonMsg::Exited { code: Some(0) } => {}
+            other => panic!("expected Exited, got {other:?}"),
+        }
+        session.detach().expect("detach");
+        server.join().expect("server thread");
+    }
+
+    #[test]
+    fn an_attach_refusal_carries_the_daemons_message() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            match ClientMsg::read(&mut r).expect("read Attach") {
+                ClientMsg::Attach { pane_id, .. } => assert_eq!(pane_id, 42),
+                other => panic!("expected Attach, got {other:?}"),
+            }
+            DaemonMsg::Error("no such pane 42".into())
+                .encode(&mut w)
+                .expect("refuse the attach");
+        });
+
+        let err = PaneSession::attach_over(client_end, 42, size(), REPLY_WAIT)
+            .expect_err("attaching to a missing pane must fail");
+        assert!(
+            err.to_string().contains("no such pane 42"),
+            "the daemon's reason was lost: {err}"
+        );
+        server.join().expect("server thread");
+    }
+
+    #[test]
+    fn an_accepted_attach_keeps_the_replay_it_peeked_at() {
+        let (client_end, server_end) = stream_pair();
+        let server = std::thread::spawn(move || {
+            let mut r = server_end.try_clone().expect("clone server end");
+            let mut w = server_end;
+            match ClientMsg::read(&mut r).expect("read Attach") {
+                ClientMsg::Attach { pane_id, .. } => assert_eq!(pane_id, 9),
+                other => panic!("expected Attach, got {other:?}"),
+            }
+            DaemonMsg::Size(size()).encode(&mut w).expect("replay size");
+            DaemonMsg::Snapshot(b"screen contents".to_vec())
+                .encode(&mut w)
+                .expect("replay snapshot");
+        });
+
+        let mut session =
+            PaneSession::attach_over(client_end, 9, size(), REPLY_WAIT).expect("attach");
+        match session.recv().expect("replayed size") {
+            DaemonMsg::Size(_) => {}
+            other => panic!("expected Size, got {other:?}"),
+        }
+        match session.recv().expect("replayed snapshot") {
+            DaemonMsg::Snapshot(bytes) => assert_eq!(bytes, b"screen contents"),
+            other => panic!("expected Snapshot, got {other:?}"),
+        }
+        server.join().expect("server thread");
+    }
+}

crates/tty7-core/src/daemon/transport.rs

@@ -63,6 +63,10 @@ mod imp_unix {
         let path = socket_path().ok_or_else(|| {
             io::Error::other("could not resolve daemon socket path (no config dir)")
         })?;
+        connect_endpoint_at(&path)
+    }
+
+    pub fn connect_endpoint_at(path: &Path) -> io::Result<Stream> {
         let stream = UnixStream::connect(path)?;
         tune(&stream);
         Ok(stream)
@@ -336,10 +340,7 @@ mod imp_windows {
         let (port, token) = read_port_file()
             .filter(|(p, _)| *p != 0)
             .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "no daemon port file"))?;
-        let mut stream = TcpStream::connect(loopback(port))?;
-        tune(&stream);
-        stream.write_all(&token)?;
-        Ok(stream)
+        connect_with_token(port, &token)
     }
 
     pub fn authenticate(stream: &mut Stream) -> io::Result<()> {
@@ -412,9 +413,26 @@ mod imp_windows {
         let (port, token) = read_port_file_named(file)
             .filter(|(p, _)| *p != 0)
             .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, format!("no {file} file")))?;
+        connect_with_token(port, &token)
+    }
+
+    pub fn connect_endpoint_at(path: &std::path::Path) -> io::Result<Stream> {
+        let contents = std::fs::read_to_string(path)?;
+        let (port, token) = parse_port_file(&contents)
+            .filter(|(p, _)| *p != 0)
+            .ok_or_else(|| {
+                io::Error::new(
+                    io::ErrorKind::NotFound,
+                    format!("no listener recorded at {}", path.display()),
+                )
+            })?;
+        connect_with_token(port, &token)
+    }
+
+    fn connect_with_token(port: u16, token: &Token) -> io::Result<Stream> {
         let mut stream = TcpStream::connect(loopback(port))?;
         tune(&stream);
-        stream.write_all(&token)?;
+        stream.write_all(token)?;
         Ok(stream)
     }
 

crates/tty7-core/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod client;
 pub mod core;
 pub mod daemon;
 pub mod host;

crates/tty7-server/tests/client_lib.rs

@@ -0,0 +1,343 @@
+use std::path::PathBuf;
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+use tty7_core::client::{ControlClient, PaneClient};
+use tty7_core::core::machine::{LayoutDelta, PaneSeed};
+use tty7_core::daemon::control::{ControlEvent, ControlHello, ControlRequest, ReplyOk, feature};
+use tty7_core::daemon::protocol::{DaemonMsg, PROTOCOL_VERSION, ShellSpec, WinSize};
+
+const READY_WITHIN: Duration = Duration::from_secs(30);
+const STREAM_WITHIN: Duration = Duration::from_secs(30);
+
+struct Daemon {
+    child: Child,
+    dir: tempfile::TempDir,
+}
+
+impl Daemon {
+    fn start() -> Daemon {
+        let dir = tempfile::TempDir::new().unwrap();
+        let child = Command::new(env!("CARGO_BIN_EXE_tty7-server"))
+            .arg("--daemon")
+            .arg("--config-dir")
+            .arg(dir.path())
+            .env("TTY7_DATA_DIR", dir.path())
+            .env("TTY7_CONTROL_SOCK", dir.path().join("control.sock"))
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .spawn()
+            .expect("start tty7-server --daemon");
+        let daemon = Daemon { child, dir };
+        daemon.await_ready();
+        daemon
+    }
+
+    fn pane_endpoint(&self) -> PathBuf {
+        let file = if cfg!(windows) {
+            "daemon.port"
+        } else {
+            "daemon.sock"
+        };
+        self.dir.path().join(file)
+    }
+
+    fn control_endpoint(&self) -> PathBuf {
+        let file = if cfg!(windows) {
+            "control.port"
+        } else {
+            "control.sock"
+        };
+        self.dir.path().join(file)
+    }
+
+    fn panes(&self) -> PaneClient {
+        PaneClient::at(self.pane_endpoint())
+    }
+
+    fn control(&self, name: &str) -> ControlClient {
+        ControlClient::connect_at(&self.control_endpoint(), &hello(name))
+            .expect("control handshake with the spawned server")
+    }
+
+    fn await_ready(&self) {
+        let deadline = Instant::now() + READY_WITHIN;
+        loop {
+            let control_up =
+                ControlClient::connect_at(&self.control_endpoint(), &hello("probe")).is_ok();
+            let panes_up = self.panes().version().is_ok();
+            if control_up && panes_up {
+                return;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "tty7-server did not open its endpoints within {READY_WITHIN:?}"
+            );
+            std::thread::sleep(Duration::from_millis(50));
+        }
+    }
+}
+
+impl Drop for Daemon {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+fn hello(name: &str) -> ControlHello {
+    ControlHello::host_rpc(name, name)
+}
+
+fn size() -> WinSize {
+    WinSize {
+        cols: 100,
+        rows: 30,
+        cell_w: 8,
+        cell_h: 16,
+    }
+}
+
+fn one_shot_shell(command: &str) -> ShellSpec {
+    if cfg!(windows) {
+        ShellSpec {
+            program: "cmd.exe".into(),
+            args: vec!["/d".into(), "/c".into(), command.into()],
+            args_are_tty7_defaults: false,
+        }
+    } else {
+        ShellSpec {
+            program: "/bin/sh".into(),
+            args: vec!["-c".into(), command.into()],
+            args_are_tty7_defaults: false,
+        }
+    }
+}
+
+fn interactive_shell() -> ShellSpec {
+    if cfg!(windows) {
+        ShellSpec {
+            program: "cmd.exe".into(),
+            args: vec!["/d".into()],
+            args_are_tty7_defaults: false,
+        }
+    } else {
+        ShellSpec {
+            program: "/bin/sh".into(),
+            args: Vec::new(),
+            args_are_tty7_defaults: false,
+        }
+    }
+}
+
+fn seed(pane: u64) -> PaneSeed {
+    PaneSeed {
+        pane,
+        cwd: Some("/home/me/proj".into()),
+        ssh_spec: None,
+        agent: None,
+    }
+}
+
+fn collect_until(
+    session: &mut tty7_core::client::PaneSession,
+    marker: &[u8],
+) -> (Vec<u8>, Option<Option<i32>>) {
+    let mut seen: Vec<u8> = Vec::new();
+    loop {
+        match session.recv() {
+            Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => {
+                seen.extend_from_slice(&bytes);
+                if windows_contain(&seen, marker) {
+                    return (seen, None);
+                }
+            }
+            Ok(DaemonMsg::Exited { code }) => return (seen, Some(code)),
+            Ok(_) => {}
+            Err(e) => panic!(
+                "pane stream ended early: {e}; saw {:?}",
+                String::from_utf8_lossy(&seen)
+            ),
+        }
+    }
+}
+
+fn drain_until_exit(session: &mut tty7_core::client::PaneSession) -> Vec<u8> {
+    let mut seen: Vec<u8> = Vec::new();
+    loop {
+        match session.recv() {
+            Ok(DaemonMsg::Output(bytes)) | Ok(DaemonMsg::Snapshot(bytes)) => {
+                seen.extend_from_slice(&bytes);
+            }
+            Ok(DaemonMsg::Exited { .. }) => return seen,
+            Ok(_) => {}
+            Err(e) => panic!(
+                "pane stream ended before Exited: {e}; saw {:?}",
+                String::from_utf8_lossy(&seen)
+            ),
+        }
+    }
+}
+
+fn windows_contain(haystack: &[u8], needle: &[u8]) -> bool {
+    haystack.windows(needle.len()).any(|w| w == needle)
+}
+
+#[test]
+fn the_pane_daemon_reports_its_dialect() {
+    let daemon = Daemon::start();
+    let version = daemon.panes().version().expect("query the version");
+    assert_eq!(version.protocol, PROTOCOL_VERSION);
+}
+
+#[test]
+fn control_requests_build_the_tree_and_events_reach_the_other_client() {
+    let daemon = Daemon::start();
+    let writer = daemon.control("writer");
+    assert!(
+        writer.hello().has_feature(feature::MACHINE_TREE),
+        "features were {:?}",
+        writer.hello().features
+    );
+
+    let watcher = daemon.control("watcher");
+    watcher
+        .request(ControlRequest::Ping)
+        .expect("the watcher is live before the writer acts");
+
+    let ws = match writer
+        .request(ControlRequest::WorkspaceCreate {
+            name: Some("api".into()),
+            workspace: None,
+        })
+        .expect("create a workspace")
+    {
+        ReplyOk::WorkspaceTree(ws) => *ws,
+        other => panic!("expected WorkspaceTree, got {other:?}"),
+    };
+    match writer
+        .request(ControlRequest::TabCreate {
+            workspace: ws.id,
+            at: None,
+            pane: seed(1),
+            tab: None,
+        })
+        .expect("create a tab")
+    {
+        ReplyOk::TabTree(_) => {}
+        other => panic!("expected TabTree, got {other:?}"),
+    }
+
+    let machine = match writer
+        .request(ControlRequest::MachineGet)
+        .expect("fetch the machine tree")
+    {
+        ReplyOk::MachineTree(m) => *m,
+        other => panic!("expected MachineTree, got {other:?}"),
+    };
+    assert_eq!(machine.workspaces.len(), 1);
+    assert_eq!(machine.workspaces[0].tabs[0].root.pane_ids(), vec![1]);
+
+    let key = ws.id.to_string();
+    let deadline = Instant::now() + STREAM_WITHIN;
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+        assert!(
+            !remaining.is_zero(),
+            "the watcher never saw the WorkspaceCreated delta for {key}"
+        );
+        match watcher.next_event(remaining) {
+            Some(ControlEvent::Layout { workspace, delta })
+                if workspace == key
+                    && matches!(delta, LayoutDelta::WorkspaceCreated { .. }) =>
+            {
+                break;
+            }
+            Some(_) => {}
+            None => {}
+        }
+    }
+}
+
+#[test]
+fn a_spawned_pane_streams_its_output_and_its_exit() {
+    let daemon = Daemon::start();
+    let mut session = daemon
+        .panes()
+        .spawn(
+            None,
+            size(),
+            Some(one_shot_shell("echo tty7_pane_roundtrip")),
+            Some("client-lib-test".into()),
+        )
+        .expect("spawn a one-shot pane");
+    assert_ne!(session.pane_id(), 0, "the daemon must name the pane");
+    session
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+
+    let seen = drain_until_exit(&mut session);
+    assert!(
+        windows_contain(&seen, b"tty7_pane_roundtrip"),
+        "output was {:?}",
+        String::from_utf8_lossy(&seen)
+    );
+}
+
+#[test]
+fn input_reaches_the_shell_and_a_reattach_replays_it() {
+    let daemon = Daemon::start();
+    let panes = daemon.panes();
+    let mut session = panes
+        .spawn(None, size(), Some(interactive_shell()), None)
+        .expect("spawn an interactive pane");
+    let pane_id = session.pane_id();
+    session
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the stream reads");
+
+    session
+        .input(b"echo tty7_attach_replay\r")
+        .expect("type into the pane");
+    let (_, exit) = collect_until(&mut session, b"tty7_attach_replay");
+    assert!(exit.is_none(), "the shell must still be running");
+    session.detach().expect("detach");
+
+    let listed = panes.list().expect("list panes");
+    let entry = listed
+        .iter()
+        .find(|p| p.pane_id == pane_id)
+        .expect("the detached pane is still listed");
+    assert!(entry.alive, "the detached pane is still alive");
+
+    let mut reattached = panes.attach(pane_id, size()).expect("reattach");
+    reattached
+        .set_recv_timeout(Some(STREAM_WITHIN))
+        .expect("bound the replay reads");
+    let (_, exit) = collect_until(&mut reattached, b"tty7_attach_replay");
+    assert!(exit.is_none(), "the replayed pane is still running");
+
+    let refused = panes
+        .attach(u64::MAX, size())
+        .expect_err("attaching to a pane that never existed must fail");
+    assert!(
+        refused.to_string().contains("no such pane"),
+        "the refusal was {refused}"
+    );
+
+    reattached.kill().expect("kill the pane");
+    let deadline = Instant::now() + STREAM_WITHIN;
+    loop {
+        let listed = panes.list().expect("list panes after the kill");
+        let gone = !listed.iter().any(|p| p.pane_id == pane_id && p.alive);
+        if gone {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "pane {pane_id} was still listed alive after Kill: {listed:?}"
+        );
+        std::thread::sleep(Duration::from_millis(50));
+    }
+}