feat(xy): CLI client commands

Replace bail!("not implemented") stubs with real RPC calls over the Unix
socket; add format::list_table for fixed-width list output.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/xy/src/cli/format.rs

@@ -0,0 +1,27 @@
+use xy_protocol::rpc::ServerSummary;
+
+pub fn list_table(rows: &[ServerSummary]) -> String {
+    let mut out = String::new();
+
+    out.push_str("NAME                STATE       PID     PORT    UPTIME    RESTARTS\n");
+
+    for r in rows {
+        let pid = r.pid.map(|p| p.to_string()).unwrap_or_else(|| "-".into());
+        let up = r
+            .uptime_secs
+            .map(|s| format!("{}s", s))
+            .unwrap_or_else(|| "-".into());
+
+        out.push_str(&format!(
+            "{:<20}{:<12}{:<8}{:<8}{:<10}{}\n",
+            r.name,
+            format!("{:?}", r.state).to_lowercase(),
+            pid,
+            r.port,
+            up,
+            r.restart_count
+        ));
+    }
+
+    out
+}

crates/xy/src/cli/mod.rs

@@ -1,24 +1,194 @@
 use crate::paths::Paths;
-use anyhow::{Result, bail};
+use anyhow::Result;
+use serde_json::json;
+use xy_ipc::{Client, ClientError};
+use xy_protocol::rpc::{
+    LogLine, LogStream, LogsParams, LogsSubscribed, ReloadResult, RestartResult, ServerSummary,
+    StartResult, StatusDetail, StopResult, methods, notifications,
+};
 
-pub async fn list(_p: Paths) -> Result<i32> {
-    bail!("not implemented")
+mod format;
+
+async fn connect(paths: &Paths) -> Result<Client> {
+    match Client::connect(&paths.socket).await {
+        Ok(c) => Ok(c),
+        Err(err) => {
+            eprintln!(
+                "xy: cannot reach daemon at {}: {err}",
+                paths.socket.display()
+            );
+            std::process::exit(2);
+        }
+    }
 }
-pub async fn status(_p: Paths, _name: String) -> Result<i32> {
-    bail!("not implemented")
+
+fn rpc_to_exit(err: &ClientError) -> i32 {
+    match err {
+        ClientError::Unreachable(_) => 2,
+        ClientError::Rpc { .. } => 1,
+        _ => 1,
+    }
 }
-pub async fn start(_p: Paths, _all: bool, _name: Option<String>) -> Result<i32> {
-    bail!("not implemented")
+
+pub async fn list(paths: Paths) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let rows: Vec<ServerSummary> = match c.call_no_params(methods::LIST).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    print!("{}", format::list_table(&rows));
+
+    Ok(0)
 }
-pub async fn stop(_p: Paths, _all: bool, _name: Option<String>) -> Result<i32> {
-    bail!("not implemented")
+
+pub async fn status(paths: Paths, name: String) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let d: StatusDetail = match c.call(methods::STATUS, &json!({"name": name})).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    println!("{:#?}", d);
+
+    Ok(0)
 }
-pub async fn restart(_p: Paths, _all: bool, _name: Option<String>) -> Result<i32> {
-    bail!("not implemented")
+
+fn name_or_all(all: bool, name: Option<String>) -> serde_json::Value {
+    if all {
+        json!({"all": true})
+    } else {
+        json!({"name": name.unwrap()})
+    }
 }
-pub async fn reload(_p: Paths) -> Result<i32> {
-    bail!("not implemented")
+
+pub async fn start(paths: Paths, all: bool, name: Option<String>) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let r: StartResult = match c.call(methods::START, &name_or_all(all, name)).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    if !r.started.is_empty() {
+        println!("started: {}", r.started.join(", "));
+    }
+
+    if !r.already_running.is_empty() {
+        println!("already running: {}", r.already_running.join(", "));
+    }
+
+    Ok(0)
 }
-pub async fn logs(_p: Paths, _name: String, _tail: Option<u32>, _follow: bool) -> Result<i32> {
-    bail!("not implemented")
+
+pub async fn stop(paths: Paths, all: bool, name: Option<String>) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let r: StopResult = match c.call(methods::STOP, &name_or_all(all, name)).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    if !r.stopped.is_empty() {
+        println!("stopped: {}", r.stopped.join(", "));
+    }
+
+    if !r.not_running.is_empty() {
+        println!("not running: {}", r.not_running.join(", "));
+    }
+
+    Ok(0)
+}
+
+pub async fn restart(paths: Paths, all: bool, name: Option<String>) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let r: RestartResult = match c.call(methods::RESTART, &name_or_all(all, name)).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    println!("restarted: {}", r.restarted.join(", "));
+
+    Ok(0)
+}
+
+pub async fn reload(paths: Paths) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let r: ReloadResult = match c.call_no_params(methods::RELOAD).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    println!("added:     {}", r.added.join(", "));
+    println!("removed:   {}", r.removed.join(", "));
+    println!("changed:   {}", r.changed.join(", "));
+    println!("unchanged: {}", r.unchanged.join(", "));
+
+    Ok(0)
+}
+
+pub async fn logs(paths: Paths, name: String, tail: Option<u32>, follow: bool) -> Result<i32> {
+    let mut c = connect(&paths).await?;
+
+    let p = LogsParams {
+        name: name.clone(),
+        tail,
+        follow,
+    };
+
+    let _sub: LogsSubscribed = match c.call(methods::LOGS, &p).await {
+        Ok(v) => v,
+        Err(err) => {
+            eprintln!("xy: {err}");
+            return Ok(rpc_to_exit(&err));
+        }
+    };
+
+    loop {
+        match c.read_notification().await {
+            Ok(None) => return Ok(0),
+            Ok(Some(n)) => match n.method.as_str() {
+                notifications::LOG => {
+                    if let Some(params) = n.params
+                        && let Ok(line) = serde_json::from_value::<LogLine>(params)
+                    {
+                        let tag = match line.stream {
+                            LogStream::Stdout => "out",
+                            LogStream::Stderr => "err",
+                        };
+
+                        println!("[{tag}] {}", line.line);
+                    }
+                }
+                notifications::LOG_END => return Ok(0),
+                _ => {}
+            },
+            Err(err) => {
+                eprintln!("xy: {err}");
+                return Ok(rpc_to_exit(&err));
+            }
+        }
+    }
 }