use crate::agent::client::AgentClient;
use crate::agent::config::AgentConfig;
use crate::agent::launchd::Launchctl;
use std::path::Path;
use std::time::{Duration, Instant};
pub fn mesh_line(m: &serde_json::Value) -> Option<String> {
let address = m["address"].as_str().unwrap_or("?");
let reach = if m["reachable"] == true {
"reachable"
} else {
"not reachable"
};
Some(match m["route"].as_str()? {
"host" => format!(
"host VPN {} ({address}) · {reach}",
m["interface"].as_str().unwrap_or("?")
),
"proxy" => format!(
"via the {} container ({address}) · {reach}",
m["interface"].as_str().unwrap_or("zakuro-wg")
),
_ => "not connected — run zc connect".to_string(),
})
}
pub fn format_status(s: &serde_json::Value) -> String {
let m = &s["this_mac"];
let mut out = format!(
" This Mac {} · broker {} · workers {}/{} running ({} busy)\n",
if m["sharing"] == true {
"sharing"
} else {
"paused"
},
m["broker"].as_str().unwrap_or("?"),
m["workers"]["running"],
m["workers"]["desired"],
m["workers"]["busy"],
);
if let Some(line) = mesh_line(&s["mesh"]) {
out += &format!(" Mesh {line}\n");
}
let price = &s["prices"]["this_mac"];
if let Some(p) = price["effective_per_hour"].as_f64() {
out += &format!(
" Price {p} cr/h ({})\n",
price["state"].as_str().unwrap_or("?")
);
}
if s["account"].is_object() {
out += &format!(
" Account {} · {} credits\n",
s["account"]["username"].as_str().unwrap_or("?"),
s["account"]["credits_balance"]
);
}
if s["earnings"].is_object() {
out += &format!(
" Earnings {} today · {} last 7 days\n",
s["earnings"]["today"], s["earnings"]["last_7d"]
);
}
if s["devices"].is_array() {
out += &format!(
" Devices {}/{} online\n",
s["devices_online"], s["devices_total"]
);
}
for p in s["problems"].as_array().into_iter().flatten() {
let hint = p["hint"]
.as_str()
.map(|h| format!(" — {h}"))
.unwrap_or_default();
out += &format!(" ! {}{hint}\n", p["message"].as_str().unwrap_or(""));
if let Some(d) = p["detail"].as_str().and_then(first_line) {
out += &format!(" {d}\n");
}
}
out
}
fn first_line(detail: &str) -> Option<String> {
let mut lines = detail.trim().lines().map(str::trim);
let first = lines.next().filter(|l| !l.is_empty())?;
Some(match lines.any(|l| !l.is_empty()) {
true => format!("{first}…"),
false => first.to_string(),
})
}
pub fn status(json: bool) -> i32 {
let Some(dir) = crate::credentials::dir().map(|d| d.join("agent")) else {
eprintln!("zc agent: no HOME or ZAKURO_HOME");
return 1;
};
let summary = AgentClient::from_dir(&dir)
.ok_or_else(|| "no ~/.zakuro/agent/agent.json".to_string())
.and_then(|c| c.summary(Duration::from_secs(2)));
match summary {
Ok(s) if json => {
println!("{}", serde_json::to_string_pretty(&s).unwrap_or_default());
0
}
Ok(s) => {
print!("{}", format_status(&s));
0
}
Err(e) => {
eprintln!("zc agent is not reachable: {e}");
1
}
}
}
pub fn install_with(
launchctl: Launchctl,
domain: &str,
exe: &Path,
home: &Path,
env: &[(&str, String)],
cfg: &AgentConfig,
) -> i32 {
match install_and_wait(launchctl, domain, exe, home, env, cfg) {
Ok(port) => {
println!(
"✓ zc agent is running on 127.0.0.1:{port} (launchd {})",
crate::agent::launchd::LABEL
);
0
}
Err(e) => {
eprintln!("{e}");
1
}
}
}
pub fn install_and_wait(
launchctl: Launchctl,
domain: &str,
exe: &Path,
home: &Path,
env: &[(&str, String)],
cfg: &AgentConfig,
) -> Result<u16, String> {
crate::agent::launchd::install(launchctl, domain, exe, home, env, &cfg.dir, cfg.port)
.map_err(|e| format!("zc agent install: {e}"))?;
wait_for_agent(&cfg.dir, Duration::from_secs(5)).ok_or_else(|| {
format!(
"zc agent was installed but did not answer within 5 s; see {}",
cfg.dir.join(crate::agent::files::LOG_FILE).display()
)
})
}
pub fn wait_for_agent(dir: &Path, timeout: Duration) -> Option<u16> {
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if let Some(c) = AgentClient::from_dir(dir) {
if c.summary(Duration::from_secs(1)).is_ok() {
return Some(c.port());
}
}
std::thread::sleep(Duration::from_millis(250));
}
None
}
#[cfg(not(target_os = "macos"))]
pub fn install() -> i32 {
eprintln!("zc agent install is not supported on this platform yet");
2
}
#[cfg(target_os = "macos")]
pub fn install() -> i32 {
let cfg = match AgentConfig::from_env() {
Ok(c) => c,
Err(e) => {
eprintln!("zc agent install: {e}");
return 1;
}
};
let home = std::path::PathBuf::from(std::env::var("HOME").unwrap_or_default());
let exe = match std::env::current_exe() {
Ok(p) => p,
Err(e) => {
eprintln!("zc agent install: {e}");
return 1;
}
};
let env = crate::agent::launchd::carried_env(|k| std::env::var(k).ok());
install_with(
&crate::agent::launchd::run_launchctl,
&crate::agent::launchd::gui_domain(),
&exe,
&home,
&env,
&cfg,
)
}
pub fn uninstall_with(launchctl: Launchctl, domain: &str, home: &Path, cfg: &AgentConfig) -> i32 {
if let Some(c) = AgentClient::from_dir(&cfg.dir) {
if c.put(
"/v1/sharing",
serde_json::json!({ "on": false }),
Duration::from_secs(5),
)
.is_ok()
{
let t = &cfg.timings;
let bound = t.drain_timeout + t.kill_grace + t.broker_stop_wait;
println!(" draining workers (up to {} s)…", bound.as_secs());
let deadline = Instant::now() + bound;
while Instant::now() < deadline {
match c.summary(Duration::from_secs(2)) {
Ok(s)
if s["this_mac"]["broker"] != "running"
&& s["this_mac"]["workers"]["running"] == 0 =>
{
break
}
Err(_) => break,
_ => std::thread::sleep(Duration::from_secs(1)),
}
}
}
}
match crate::agent::launchd::uninstall(launchctl, domain, home) {
Ok(()) => {
println!("✓ zc agent uninstalled ({} kept)", cfg.dir.display());
0
}
Err(e) => {
eprintln!("zc agent uninstall: {e}");
1
}
}
}
#[cfg(not(target_os = "macos"))]
pub fn uninstall() -> i32 {
eprintln!("zc agent uninstall is not supported on this platform yet");
2
}
#[cfg(target_os = "macos")]
pub fn uninstall() -> i32 {
let cfg = match AgentConfig::from_env() {
Ok(c) => c,
Err(e) => {
eprintln!("zc agent uninstall: {e}");
return 1;
}
};
let home = std::path::PathBuf::from(std::env::var("HOME").unwrap_or_default());
uninstall_with(
&crate::agent::launchd::run_launchctl,
&crate::agent::launchd::gui_domain(),
&home,
&cfg,
)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::agent::client::tests::AgentGuard;
use crate::agent::config::AgentConfig;
use crate::agent::files::tests::tmp;
#[test]
fn status_prints_the_essentials_and_every_problem() {
let fixture: serde_json::Value =
serde_json::from_str(include_str!("../../docs/agent-api/summary.v1.json")).unwrap();
let out = super::format_status(&fixture);
assert!(
out.contains("sharing · broker running · workers 2/2 running (1 busy)"),
"{out}"
);
assert!(out.contains("18 cr/h (set)"), "{out}");
assert!(out.contains("jean · 1240.5 credits"), "{out}");
assert!(out.contains("12.4 today · 88.1 last 7 days"), "{out}");
assert!(out.contains("2/3 online"), "{out}");
assert!(
out.contains(" Mesh via the zakuro-wg container (10.13.13.7) · reachable\n"),
"{out}"
);
assert!(
out.contains("! uv is not installed — brew install uv"),
"{out}"
);
}
fn fixture() -> serde_json::Value {
serde_json::from_str(include_str!("../../docs/agent-api/summary.v1.json")).unwrap()
}
#[test]
fn a_problem_shows_its_next_step_under_it() {
let mut s = fixture();
s["problems"] = serde_json::json!([
{
"code": "hub_unreachable",
"message": "Can't reach the Zakuro hub",
"hint": "No cached values yet",
"detail": "this host may answer only over a VPN tunnel: check that the tunnel carrying it is up",
},
{
"code": "worker_crashloop",
"message": "A worker keeps crashing",
"hint": "See ~/.zakuro/agent/workers/",
"detail": "first log line\nsecond log line\nthird log line",
},
]);
let out = format_status(&s);
assert!(
out.contains(" ! Can't reach the Zakuro hub — No cached values yet\n this host may answer only over a VPN tunnel: check that the tunnel carrying it is up\n"),
"{out}"
);
assert!(out.contains(" first log line…\n"), "{out}");
assert!(
!out.contains("second log line"),
"a log tail is not the status screen: {out}"
);
}
#[test]
fn with_nothing_cached_no_device_account_or_earnings_line_is_printed() {
let mut s = fixture();
s["account"] = serde_json::Value::Null;
s["earnings"] = serde_json::Value::Null;
s["devices"] = serde_json::Value::Null;
s["devices_online"] = serde_json::json!(0);
s["devices_total"] = serde_json::json!(0);
s["prices"]["this_mac"] = serde_json::json!({
"state": "inherited", "per_hour": null, "effective_per_hour": null
});
s["problems"] = serde_json::json!([{
"code": "hub_unreachable",
"message": "Can't reach the Zakuro hub",
"hint": "No cached values yet: showing this Mac only",
"detail": null,
}]);
let out = format_status(&s);
assert!(
out.contains(" This Mac sharing · broker running"),
"{out}"
);
assert!(!out.contains("Devices"), "no zeros to read as none: {out}");
assert!(!out.contains("Account"), "{out}");
assert!(!out.contains("Earnings"), "{out}");
assert!(!out.contains("Price"), "{out}");
assert!(
out.contains(
"! Can't reach the Zakuro hub — No cached values yet: showing this Mac only\n"
),
"{out}"
);
let mut answered = fixture();
answered["devices"] = serde_json::json!([]);
answered["devices_online"] = serde_json::json!(0);
answered["devices_total"] = serde_json::json!(0);
assert!(
format_status(&answered).contains(" Devices 0/0 online\n"),
"the hub answered and listed none: that is a real 0/0"
);
}
#[test]
fn status_prints_the_mesh_route() {
use serde_json::json;
let line = |v: serde_json::Value| super::mesh_line(&v);
assert_eq!(
line(json!({"route": "host", "address": "10.13.13.7", "interface": "utun4", "reachable": true})).as_deref(),
Some("host VPN utun4 (10.13.13.7) · reachable")
);
assert_eq!(
line(json!({"route": "proxy", "address": "10.13.13.7", "interface": "zakuro-wg", "reachable": true})).as_deref(),
Some("via the zakuro-wg container (10.13.13.7) · reachable")
);
assert_eq!(
line(json!({"route": "proxy", "address": "10.13.13.7", "interface": "zakuro-wg", "reachable": false})).as_deref(),
Some("via the zakuro-wg container (10.13.13.7) · not reachable")
);
assert_eq!(
line(json!({"route": "none", "address": null, "interface": null, "reachable": false}))
.as_deref(),
Some("not connected — run zc connect")
);
assert_eq!(
line(serde_json::Value::Null),
None,
"an agent from before the mesh route was reported sends no mesh"
);
}
struct FakeLaunchctl {
calls: std::sync::Mutex<Vec<Vec<String>>>,
}
impl FakeLaunchctl {
fn new() -> Self {
Self {
calls: std::sync::Mutex::new(vec![]),
}
}
fn call(&self, args: &[&str]) -> std::io::Result<std::process::ExitStatus> {
self.calls
.lock()
.unwrap()
.push(args.iter().map(|s| s.to_string()).collect());
std::process::Command::new("true").status()
}
fn calls(&self) -> Vec<Vec<String>> {
self.calls.lock().unwrap().clone()
}
}
fn running_agent() -> (AgentConfig, AgentGuard) {
let free = || {
std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
};
let sleeper = vec![
"sh".to_string(),
"-c".to_string(),
"exec sleep 30".to_string(),
];
let mut cfg = AgentConfig::for_dirs(tmp("cli-uninstall"));
cfg.port = 0;
cfg.broker_port = free();
cfg.max_workers = 3;
cfg.broker_argv = sleeper.clone();
cfg.worker_template = sleeper;
cfg.worker_template_overridden = true;
cfg.timings.shutdown_wait = std::time::Duration::from_secs(2);
cfg.timings.drain_timeout = std::time::Duration::from_secs(1);
cfg.timings.kill_grace = std::time::Duration::from_secs(0);
cfg.timings.broker_stop_wait = std::time::Duration::from_secs(0);
let (_, _, stop, thread) =
crate::agent::run::start_background(cfg.clone()).expect("agent starts");
(cfg, AgentGuard::new(stop, thread))
}
#[test]
fn uninstall_turns_sharing_off_through_the_client_before_booting_out() {
let (cfg, guard) = running_agent();
let client =
crate::agent::client::AgentClient::from_dir(&cfg.dir).expect("agent.json written");
let t = std::time::Duration::from_secs(2);
client
.put("/v1/sharing", serde_json::json!({ "on": true }), t)
.unwrap();
assert_eq!(client.summary(t).unwrap()["this_mac"]["sharing"], true);
let home = tmp("cli-uninstall-home");
std::fs::create_dir_all(&home).unwrap();
let fake = FakeLaunchctl::new();
let sharing_at_bootout = std::sync::Mutex::new(vec![]);
let launchctl = |args: &[&str]| {
if args.first() == Some(&"bootout") {
sharing_at_bootout.lock().unwrap().push(
client
.summary(t)
.ok()
.map(|s| s["this_mac"]["sharing"].clone()),
);
}
fake.call(args)
};
let code = uninstall_with(&launchctl, "gui/501", &home, &cfg);
assert_eq!(code, 0);
assert_eq!(
*sharing_at_bootout.lock().unwrap(),
vec![Some(serde_json::json!(false))],
"one bootout, and sharing was already off when it ran"
);
assert_eq!(
fake.calls().last().unwrap(),
&vec!["bootout", "gui/501/ai.zakuro.agent"]
);
drop(guard);
}
#[test]
fn install_reports_the_port_the_launched_agent_answers_on() {
let home = tmp("cli-install-home");
std::fs::create_dir_all(&home).unwrap();
let launched = crate::agent::client::tests::sleeper_cfg("cli-install");
let mut asked = launched.clone();
asked.port = 1; let started = std::sync::Mutex::new(None);
let launchctl = |args: &[&str]| {
if args.first() == Some(&"bootstrap") {
let (_, _, stop, thread) =
crate::agent::run::start_background(launched.clone()).expect("agent starts");
*started.lock().unwrap() = Some(AgentGuard::new(stop, thread));
}
let ok = args.first() != Some(&"print");
std::process::Command::new(if ok { "true" } else { "false" }).status()
};
let port = install_and_wait(
&launchctl,
"gui/501",
std::path::Path::new("/usr/local/bin/zc"),
&home,
&[],
&asked,
)
.expect("the launched agent answers");
let bound = crate::agent::files::load_json::<crate::agent::files::AgentFile>(
&asked.dir.join(crate::agent::files::AGENT_FILE),
)
.unwrap()
.port;
assert_eq!(port, bound, "the port the launched agent answers on");
assert_ne!(port, asked.port, "not the port the config asked for");
drop(started);
}
}